SAMZA-1948: Updated hybrid table descriptors to take underlying table descriptors

Hybrid Table Descriptors, e.g. CachingTableDescriptor, should take both tables as a constructor param instead of using withXYZ methods. This is better because they are required for the hybrid descriptor to work, and it helps with type inference for the descriptor.

Author: Wei Song <wsong@linkedin.com>

Reviewers: Peng Du <pdu@linkedin.com>

Closes apache#706 from weisong44/SAMZA-1948 and squashes the following commits:

53444419 [Wei Song] Updated based on review comments
39d9ab00 [Wei Song] SAMZA-1948 Updated hybrid table descriptors to take underlying table descriptors
a56c28d [Wei Song] Merge remote-tracking branch 'upstream/master'
097958c [Wei Song] Merge remote-tracking branch 'upstream/master'
05822f0 [Wei Song] Merge remote-tracking branch 'upstream/master'
f748050 [Wei Song] Merge remote-tracking branch 'upstream/master'
7706ab1 [Wei Song] Merge remote-tracking branch 'upstream/master'
f5731b1 [Wei Song] Merge remote-tracking branch 'upstream/master'
1e5de45 [Wei Song] Merge remote-tracking branch 'upstream/master'
c85604e [Wei Song] Merge remote-tracking branch 'upstream/master'
242d844 [Wei Song] Merge remote-tracking branch 'upstream/master'
ec7d840 [Wei Song] Merge remote-tracking branch 'upstream/master'
e19b4dc [Wei Song] Merge remote-tracking branch 'upstream/master'
8ee7844 [Wei Song] Merge remote-tracking branch 'upstream/master'
1c6a2ea [Wei Song] Merge remote-tracking branch 'upstream/master'
a6c94ad [Wei Song] Merge remote-tracking branch 'upstream/master'
41299b5 [Wei Song] Merge remote-tracking branch 'upstream/master'
239a095 [Wei Song] Merge remote-tracking branch 'upstream/master'
eca0020 [Wei Song] Merge remote-tracking branch 'upstream/master'
5156239 [Wei Song] Merge remote-tracking branch 'upstream/master'
de708f5 [Wei Song] Merge remote-tracking branch 'upstream/master'
df2f8d7 [Wei Song] Merge remote-tracking branch 'upstream/master'
f28b491 [Wei Song] Merge remote-tracking branch 'upstream/master'
4782c61 [Wei Song] Merge remote-tracking branch 'upstream/master'
0440f75 [Wei Song] Merge remote-tracking branch 'upstream/master'
aae0f38 [Wei Song] Merge remote-tracking branch 'upstream/master'
a15a7c9 [Wei Song] Merge remote-tracking branch 'upstream/master'
5cbf9af [Wei Song] Merge remote-tracking branch 'upstream/master'
3f7ed71 [Wei Song] Added self to committer list

Author: weisong44

samza-api/src/main/java/org/apache/samza/table/TableDescriptorsProvider.java

@@ -91,7 +91,7 @@
 public interface TableDescriptorsProvider {
   /**
    * Constructs instances of the table descriptors
-   * @param config
+   * @param config the job config
    * @return list of table descriptors
    */
   List<TableDescriptor> getTableDescriptors(Config config);

samza-core/src/main/java/org/apache/samza/table/caching/CachingTableDescriptor.java

@@ -47,10 +47,30 @@ public class CachingTableDescriptor<K, V> extends BaseHybridTableDescriptor<K, V
   private boolean isWriteAround;
 
   /**
-   * {@inheritDoc}
+   * Constructs a table descriptor instance with internal cache
+   *
+   * @param tableId Id of the table, it must conform to pattern { @literal [\\d\\w-_]+ }
+   * @param table target table descriptor
    */
-  public CachingTableDescriptor(String tableId) {
+  public CachingTableDescriptor(String tableId, TableDescriptor<K, V, ?> table) {
     super(tableId);
+    this.table = table;
+  }
+
+  /**
+   * Constructs a table descriptor instance and specify a cache (as Table descriptor)
+   * to be used for caching. Cache get is not synchronized with put for better parallelism
+   * in the read path of {@link CachingTable}. As such, cache table implementation is
+   * expected to be thread-safe for concurrent accesses.
+   *
+   * @param tableId Id of the table, it must conform to pattern { @literal [\\d\\w-_]+ }
+   * @param table target table descriptor
+   * @param cache cache table descriptor
+   */
+  public CachingTableDescriptor(String tableId, TableDescriptor<K, V, ?> table,
+      TableDescriptor<K, V, ?> cache) {
+    this(tableId, table);
+    this.cache = cache;
   }
 
   @Override
@@ -87,29 +107,6 @@ public TableSpec getTableSpec() {
     return new TableSpec(tableId, serde, CachingTableProviderFactory.class.getName(), tableSpecConfig);
   }
 
-  /**
-   * Specify a cache (as Table descriptor) to be used for caching.
-   * Cache get is not synchronized with put for better parallelism in the read path
-   * of {@link CachingTable}. As such, cache table implementation is expected to be
-   * thread-safe for concurrent accesses.
-   * @param cache cache table descriptor
-   * @return this descriptor
-   */
-  public CachingTableDescriptor withCache(TableDescriptor<K, V, ?> cache) {
-    this.cache = cache;
-    return this;
-  }
-
-  /**
-   * Specify the target table descriptor for the actual table input/output.
-   * @param table the target table descriptor
-   * @return this descriptor
-   */
-  public CachingTableDescriptor withTable(TableDescriptor<K, V, ?> table) {
-    this.table = table;
-    return this;
-  }
-
   /**
    * Specify the TTL for each read access, ie. record is expired after
    * the TTL duration since last read access of each key.

samza-core/src/test/java/org/apache/samza/table/caching/TestCachingTable.java

@@ -82,14 +82,15 @@ public void testSerializeWithCacheInstance() {
   }
 
   private void doTestSerialize(TableDescriptor cache) {
-    CachingTableDescriptor desc = new CachingTableDescriptor("1");
-    desc.withTable(createDummyTableDescriptor("2"));
+    CachingTableDescriptor desc;
+    TableDescriptor table = createDummyTableDescriptor("2");
     if (cache == null) {
+      desc = new CachingTableDescriptor("1", table);
       desc.withReadTtl(Duration.ofMinutes(3));
       desc.withWriteTtl(Duration.ofMinutes(3));
       desc.withCacheSize(1000);
     } else {
-      desc.withCache(cache);
+      desc = new CachingTableDescriptor("1", table, cache);
     }
 
     desc.withWriteAround();
@@ -150,9 +151,9 @@ private void initTables(ReadableTable ... tables) {
   }
 
   private void doTestCacheOps(boolean isWriteAround) {
-    CachingTableDescriptor desc = new CachingTableDescriptor("1");
-    desc.withTable(createDummyTableDescriptor("realTable"));
-    desc.withCache(createDummyTableDescriptor("cacheTable"));
+    CachingTableDescriptor desc = new CachingTableDescriptor("1",
+        createDummyTableDescriptor("realTable"),
+        createDummyTableDescriptor("cacheTable"));
     if (isWriteAround) {
       desc.withWriteAround();
     }

samza-test/src/test/java/org/apache/samza/test/table/TestRemoteTable.java

@@ -144,17 +144,17 @@ public boolean isRetriable(Throwable exception) {
   }
 
   private <K, V> Table<KV<K, V>> getCachingTable(TableDescriptor<K, V, ?> actualTableDesc, boolean defaultCache, String id, StreamApplicationDescriptor appDesc) {
-    CachingTableDescriptor<K, V> cachingDesc = new CachingTableDescriptor<>("caching-table-" + id);
+    CachingTableDescriptor<K, V> cachingDesc;
     if (defaultCache) {
+      cachingDesc = new CachingTableDescriptor<>("caching-table-" + id, actualTableDesc);
       cachingDesc.withReadTtl(Duration.ofMinutes(5));
       cachingDesc.withWriteTtl(Duration.ofMinutes(5));
     } else {
       GuavaCacheTableDescriptor<K, V> guavaTableDesc = new GuavaCacheTableDescriptor<>("guava-table-" + id);
       guavaTableDesc.withCache(CacheBuilder.newBuilder().expireAfterAccess(5, TimeUnit.MINUTES).build());
-      cachingDesc.withCache(guavaTableDesc);
+      cachingDesc = new CachingTableDescriptor<>("caching-table-" + id, actualTableDesc, guavaTableDesc);
     }
 
-    cachingDesc.withTable(actualTableDesc);
     return appDesc.getTable(cachingDesc);
   }