Merge remote-tracking branch 'upstream/master'

weisong44 · weisong44 · commit a6c94add7724 · 2018-09-20T10:39:31.000-07:00
diff --git a/samza-api/src/main/java/org/apache/samza/operators/TableDescriptor.java b/samza-api/src/main/java/org/apache/samza/operators/TableDescriptor.java
@@ -19,7 +19,6 @@
 package org.apache.samza.operators;
 
 import org.apache.samza.annotation.InterfaceStability;
-import org.apache.samza.serializers.KVSerde;
 
 /**
  * User facing class to collect metadata that fully describes a
@@ -30,8 +29,8 @@
  *
  * <pre>
  * {@code
- * TableDescriptor<Integer, String, ?> tableDesc = new RocksDbTableDescriptor("tbl")
- *     .withSerde(KVSerde.of(new IntegerSerde(), new StringSerde("UTF-8")))
+ * TableDescriptor<Integer, String, ?> tableDesc = new RocksDbTableDescriptor("tbl",
+ *         KVSerde.of(new IntegerSerde(), new StringSerde("UTF-8")))
  *     .withBlockSize(1024)
  *     .withConfig("some-key", "some-value");
  * }
@@ -54,14 +53,6 @@ public interface TableDescriptor<K, V, D extends TableDescriptor<K, V, D>> {
    */
   String getTableId();
 
-  /**
-   * Set the Serde for this table
-   * @param serde the serde
-   * @return this table descriptor instance
-   * @throws IllegalArgumentException if null is provided
-   */
-  D withSerde(KVSerde<K, V> serde);
-
   /**
    * Add a configuration entry for the table
    * @param key the key
diff --git a/samza-api/src/main/java/org/apache/samza/table/TableDescriptorsProvider.java b/samza-api/src/main/java/org/apache/samza/table/TableDescriptorsProvider.java
@@ -44,13 +44,11 @@
  *   public List<TableDescriptor> getTableDescriptors() {
  *     List<TableDescriptor> tableDescriptors = new ArrayList<>();
  *     final TableReadFunction readRemoteTableFn = new MyStoreReadFunction();
- *     tableDescriptors.add(new RemoteTableDescriptor<>("remote-table-1")
- *       .withReadFunction(readRemoteTableFn)
- *       .withSerde(KVSerde.of(new StringSerde(), new StringSerde())));
+ *     tableDescriptors.add(new RemoteTableDescriptor<>("remote-table-1", KVSerde.of(new StringSerde(), new StringSerde()))
+ *       .withReadFunction(readRemoteTableFn);
  *
- *     tableDescriptors.add(new RocksDbTableDescriptor("local-table-1")
+ *     tableDescriptors.add(new RocksDbTableDescriptor("local-table-1", KVSerde.of(new LongSerde(), new StringSerde<>()))
  *       .withBlockSize(4096)
- *       .withSerde(KVSerde.of(new LongSerde(), new StringSerde<>())));
  *       .withConfig("some-key", "some-value");
  *     return tableDescriptors;
  *   }
diff --git a/samza-core/src/main/java/org/apache/samza/operators/BaseTableDescriptor.java b/samza-core/src/main/java/org/apache/samza/operators/BaseTableDescriptor.java
@@ -51,18 +51,19 @@ protected BaseTableDescriptor(String tableId) {
     this.tableId = tableId;
   }
 
-  @Override
-  public D withConfig(String key, String value) {
-    config.put(key, value);
-    return (D) this;
+  /**
+   * Constructs a table descriptor instance
+   * @param tableId Id of the table, it must confirm to pattern { @literal [\\d\\w-_]+ }
+   * @param serde the serde for key and value
+   */
+  protected BaseTableDescriptor(String tableId, KVSerde<K, V> serde) {
+    this.tableId = tableId;
+    this.serde = serde;
   }
 
   @Override
-  public D withSerde(KVSerde<K, V> serde) {
-    if (serde == null) {
-      throw new IllegalArgumentException("Serde cannot be null");
-    }
-    this.serde = serde;
+  public D withConfig(String key, String value) {
+    config.put(key, value);
     return (D) this;
   }
 
diff --git a/samza-core/src/main/java/org/apache/samza/table/remote/RemoteTableDescriptor.java b/samza-core/src/main/java/org/apache/samza/table/remote/RemoteTableDescriptor.java
@@ -23,6 +23,7 @@
 import java.util.Map;
 
 import org.apache.samza.operators.BaseTableDescriptor;
+import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.table.TableSpec;
 import org.apache.samza.table.retry.TableRetryPolicy;
 import org.apache.samza.table.utils.SerdeUtils;
@@ -79,12 +80,22 @@ public class RemoteTableDescriptor<K, V> extends BaseTableDescriptor<K, V, Remot
   private int asyncCallbackPoolSize = -1;
 
   /**
-   * {@inheritDoc}
+   * Constructs a table descriptor instance
+   * @param tableId Id of the table, it must confirm to pattern { @literal [\\d\\w-_]+ }
    */
   public RemoteTableDescriptor(String tableId) {
     super(tableId);
   }
 
+  /**
+   * Constructs a table descriptor instance
+   * @param tableId Id of the table, it must confirm to pattern { @literal [\\d\\w-_]+ }
+   * @param serde the serde for key and value
+   */
+  public RemoteTableDescriptor(String tableId, KVSerde<K, V> serde) {
+    super(tableId, serde);
+  }
+
   @Override
   public TableSpec getTableSpec() {
     validate();
diff --git a/samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/InMemoryTableDescriptor.java b/samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/InMemoryTableDescriptor.java
@@ -21,6 +21,7 @@
 import java.util.HashMap;
 import java.util.Map;
 
+import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.storage.kv.BaseLocalStoreBackedTableDescriptor;
 import org.apache.samza.table.TableSpec;
 
@@ -34,12 +35,22 @@
 public class InMemoryTableDescriptor<K, V> extends BaseLocalStoreBackedTableDescriptor<K, V, InMemoryTableDescriptor<K, V>> {
 
   /**
-   * {@inheritDoc}
+   * Constructs a table descriptor instance
+   * @param tableId Id of the table, it must confirm to pattern { @literal [\\d\\w-_]+ }
    */
   public InMemoryTableDescriptor(String tableId) {
     super(tableId);
   }
 
+  /**
+   * Constructs a table descriptor instance
+   * @param tableId Id of the table, it must confirm to pattern { @literal [\\d\\w-_]+ }
+   * @param serde the serde for key and value
+   */
+  public InMemoryTableDescriptor(String tableId, KVSerde<K, V> serde) {
+    super(tableId, serde);
+  }
+
   @Override
   protected void generateTableSpecConfig(Map<String, String> tableSpecConfig) {
     super.generateTableSpecConfig(tableSpecConfig);
diff --git a/samza-kv-inmemory/src/test/java/org/apache/samza/storage/kv/inmemory/TestInMemoryTableDescriptor.java b/samza-kv-inmemory/src/test/java/org/apache/samza/storage/kv/inmemory/TestInMemoryTableDescriptor.java
@@ -31,8 +31,8 @@ public class TestInMemoryTableDescriptor {
   @Test
   public void testTableSpec() {
 
-    TableSpec tableSpec = new InMemoryTableDescriptor<Integer, String>("1")
-        .withSerde(KVSerde.of(new IntegerSerde(), new StringSerde()))
+    TableSpec tableSpec = new InMemoryTableDescriptor("1",
+            KVSerde.of(new IntegerSerde(), new StringSerde()))
         .withConfig("inmemory.abc", "xyz")
         .getTableSpec();
 
diff --git a/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/RocksDbTableDescriptor.java b/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/RocksDbTableDescriptor.java
@@ -21,6 +21,7 @@
 import java.util.HashMap;
 import java.util.Map;
 
+import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.table.TableSpec;
 
 
@@ -57,12 +58,22 @@ public class RocksDbTableDescriptor<K, V> extends BaseLocalStoreBackedTableDescr
   private String compactionStyle;
 
   /**
-   * {@inheritDoc}
+   * Constructs a table descriptor instance
+   * @param tableId Id of the table, it must confirm to pattern { @literal [\\d\\w-_]+ }
    */
   public RocksDbTableDescriptor(String tableId) {
     super(tableId);
   }
 
+  /**
+   * Constructs a table descriptor instance
+   * @param tableId Id of the table, it must confirm to pattern { @literal [\\d\\w-_]+ }
+   * @param serde the serde for key and value
+   */
+  public RocksDbTableDescriptor(String tableId, KVSerde<K, V> serde) {
+    super(tableId, serde);
+  }
+
   /**
    * Refer to <code>stores.store-name.write.batch.size</code> in Samza configuration guide
    * @param writeBatchSize write batch size
diff --git a/samza-kv-rocksdb/src/test/java/org/apache/samza/storage/kv/TestRocksDbTableDescriptor.java b/samza-kv-rocksdb/src/test/java/org/apache/samza/storage/kv/TestRocksDbTableDescriptor.java
@@ -37,8 +37,8 @@ public void testMinimal() {
 
   @Test
   public void testSerde() {
-    TableSpec tableSpec = new RocksDbTableDescriptor<Integer, String>("1")
-        .withSerde(KVSerde.of(new IntegerSerde(), new StringSerde()))
+    TableSpec tableSpec = new RocksDbTableDescriptor("1",
+            KVSerde.of(new IntegerSerde(), new StringSerde()))
         .getTableSpec();
     Assert.assertNotNull(tableSpec.getSerde());
     Assert.assertEquals(tableSpec.getSerde().getKeySerde().getClass(), IntegerSerde.class);
@@ -48,8 +48,8 @@ public void testSerde() {
   @Test
   public void testTableSpec() {
 
-    TableSpec tableSpec = new RocksDbTableDescriptor<Integer, String>("1")
-        .withSerde(KVSerde.of(new IntegerSerde(), new StringSerde()))
+    TableSpec tableSpec = new RocksDbTableDescriptor<Integer, String>("1",
+            KVSerde.of(new IntegerSerde(), new StringSerde()))
         .withBlockSize(1)
         .withCacheSize(2)
         .withCompactionStyle("fifo")
@@ -85,8 +85,7 @@ public void testTableSpec() {
   @Test
   public void testTableSpecWithChangelogEnabled() {
 
-    TableSpec tableSpec = new RocksDbTableDescriptor<Integer, String>("1")
-        .withSerde(KVSerde.of(new IntegerSerde(), new StringSerde()))
+    TableSpec tableSpec = new RocksDbTableDescriptor("1", KVSerde.of(new IntegerSerde(), new StringSerde()))
         .withChangelogStream("changelog-$tream")
         .withChangelogReplicationFactor(10)
         .getTableSpec();
diff --git a/samza-kv/src/main/java/org/apache/samza/storage/kv/BaseLocalStoreBackedTableDescriptor.java b/samza-kv/src/main/java/org/apache/samza/storage/kv/BaseLocalStoreBackedTableDescriptor.java
@@ -24,6 +24,7 @@
 import java.util.Map;
 
 import org.apache.samza.operators.BaseTableDescriptor;
+import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.storage.SideInputsProcessor;
 
 
@@ -49,13 +50,21 @@ abstract public class BaseLocalStoreBackedTableDescriptor<K, V, D extends BaseLo
 
   /**
    * Constructs a table descriptor instance
-   *
-   * @param tableId Id of the table
+   * @param tableId Id of the table, it must confirm to pattern { @literal [\\d\\w-_]+ }
    */
   public BaseLocalStoreBackedTableDescriptor(String tableId) {
     super(tableId);
   }
 
+  /**
+   * Constructs a table descriptor instance
+   * @param tableId Id of the table, it must confirm to pattern { @literal [\\d\\w-_]+ }
+   * @param serde the serde for key and value
+   */
+  public BaseLocalStoreBackedTableDescriptor(String tableId, KVSerde<K, V> serde) {
+    super(tableId, serde);
+  }
+
   public D withSideInputs(List<String> sideInputs) {
     this.sideInputs = sideInputs;
     // Disable changelog
diff --git a/samza-sql/src/main/java/org/apache/samza/sql/impl/ConfigBasedIOResolverFactory.java b/samza-sql/src/main/java/org/apache/samza/sql/impl/ConfigBasedIOResolverFactory.java
@@ -100,10 +100,9 @@ private SqlIOConfig fetchSystemInfo(String name) {
 
       TableDescriptor tableDescriptor = null;
       if (isTable) {
-        tableDescriptor = new RocksDbTableDescriptor("InputTable-" + name)
-            .withSerde(KVSerde.of(
-                new JsonSerdeV2<>(SamzaSqlCompositeKey.class),
-                new JsonSerdeV2<>(SamzaSqlRelMessage.class)));
+        tableDescriptor = new RocksDbTableDescriptor("InputTable-" + name, KVSerde.of(
+            new JsonSerdeV2<>(SamzaSqlCompositeKey.class),
+            new JsonSerdeV2<>(SamzaSqlRelMessage.class)));
       }
 
       return new SqlIOConfig(systemName, streamName, fetchSystemConfigs(systemName), tableDescriptor);
diff --git a/samza-sql/src/test/java/org/apache/samza/sql/testutil/TestIOResolverFactory.java b/samza-sql/src/test/java/org/apache/samza/sql/testutil/TestIOResolverFactory.java
@@ -202,10 +202,9 @@ private SqlIOConfig fetchIOInfo(String ioName, boolean isSink) {
             tableDescriptor = new TestTableDescriptor(TEST_TABLE_ID + tableDescMap.size());
           } else {
             String tableId = "InputTable-" + ioName.replace(".", "-").replace("$", "-");
-            tableDescriptor = new RocksDbTableDescriptor(tableId)
-                .withSerde(KVSerde.of(
-                    new JsonSerdeV2<>(SamzaSqlCompositeKey.class),
-                    new JsonSerdeV2<>(SamzaSqlRelMessage.class)));
+            tableDescriptor = new RocksDbTableDescriptor(tableId, KVSerde.of(
+                new JsonSerdeV2<>(SamzaSqlCompositeKey.class),
+                new JsonSerdeV2<>(SamzaSqlRelMessage.class)));
           }
           tableDescMap.put(ioName, tableDescriptor);
         }
diff --git a/samza-test/src/test/java/org/apache/samza/test/table/TestLocalTable.java b/samza-test/src/test/java/org/apache/samza/test/table/TestLocalTable.java
@@ -96,8 +96,8 @@ public void testSendTo() throws  Exception {
 
     final StreamApplication app = appDesc -> {
 
-      Table<KV<Integer, Profile>> table = appDesc.getTable(new InMemoryTableDescriptor("t1")
-          .withSerde(KVSerde.of(new IntegerSerde(), new ProfileJsonSerde())));
+      Table<KV<Integer, Profile>> table = appDesc.getTable(new InMemoryTableDescriptor("t1",
+          KVSerde.of(new IntegerSerde(), new ProfileJsonSerde())));
       DelegatingSystemDescriptor ksd = new DelegatingSystemDescriptor("test");
       GenericInputDescriptor<Profile> isd = ksd.getInputDescriptor("Profile", new NoOpSerde<>());
 
@@ -134,7 +134,7 @@ void runTest() {
       final StreamApplication app = appDesc -> {
 
         Table<KV<Integer, Profile>> table = appDesc.getTable(
-            new InMemoryTableDescriptor("t1").withSerde(KVSerde.of(new IntegerSerde(), new ProfileJsonSerde())));
+            new InMemoryTableDescriptor("t1", KVSerde.of(new IntegerSerde(), new ProfileJsonSerde())));
         DelegatingSystemDescriptor ksd = new DelegatingSystemDescriptor("test");
         GenericInputDescriptor<Profile> profileISD = ksd.getInputDescriptor("Profile", new NoOpSerde<>());
         appDesc.getInputStream(profileISD)
@@ -209,8 +209,7 @@ void runTest() {
 
       final StreamApplication app = appDesc -> {
 
-        Table<KV<Integer, Profile>> profileTable = appDesc.getTable(new InMemoryTableDescriptor("t1")
-            .withSerde(profileKVSerde));
+        Table<KV<Integer, Profile>> profileTable = appDesc.getTable(new InMemoryTableDescriptor("t1", profileKVSerde));
 
         DelegatingSystemDescriptor ksd = new DelegatingSystemDescriptor("test");
         GenericInputDescriptor<Profile> profileISD1 = ksd.getInputDescriptor("Profile1", new NoOpSerde<>());
diff --git a/samza-test/src/test/java/org/apache/samza/test/table/TestLocalTableWithSideInputs.java b/samza-test/src/test/java/org/apache/samza/test/table/TestLocalTableWithSideInputs.java
@@ -143,8 +143,7 @@ public void describe(StreamApplicationDescriptor appDesc) {
     }
 
     protected TableDescriptor<Integer, Profile, ?> getTableDescriptor() {
-      return new InMemoryTableDescriptor<Integer, Profile>(PROFILE_TABLE)
-          .withSerde(KVSerde.of(new IntegerSerde(), new ProfileJsonSerde()))
+      return new InMemoryTableDescriptor(PROFILE_TABLE, KVSerde.of(new IntegerSerde(), new ProfileJsonSerde()))
           .withSideInputs(ImmutableList.of(PROFILE_STREAM))
           .withSideInputsProcessor((msg, store) -> {
               Profile profile = (Profile) msg.getMessage();
@@ -158,8 +157,7 @@ public void describe(StreamApplicationDescriptor appDesc) {
   static class DurablePageViewProfileJoin extends PageViewProfileJoin {
     @Override
     protected TableDescriptor<Integer, Profile, ?> getTableDescriptor() {
-      return new RocksDbTableDescriptor<Integer, Profile>(PROFILE_TABLE)
-          .withSerde(KVSerde.of(new IntegerSerde(), new ProfileJsonSerde()))
+      return new RocksDbTableDescriptor(PROFILE_TABLE, KVSerde.of(new IntegerSerde(), new ProfileJsonSerde()))
           .withSideInputs(ImmutableList.of(PROFILE_STREAM))
           .withSideInputsProcessor((msg, store) -> {
               TestTableData.Profile profile = (TestTableData.Profile) msg.getMessage();
diff --git a/samza-test/src/test/java/org/apache/samza/test/table/TestTableDescriptorsProvider.java b/samza-test/src/test/java/org/apache/samza/test/table/TestTableDescriptorsProvider.java
@@ -137,13 +137,11 @@ public List<TableDescriptor> getTableDescriptors(Config config) {
       final RateLimiter readRateLimiter = mock(RateLimiter.class);
       final MyReadFunction readFn = new MyReadFunction();
 
-      tableDescriptors.add(new RemoteTableDescriptor<>("remote-table-1")
+      tableDescriptors.add(new RemoteTableDescriptor<>("remote-table-1", KVSerde.of(new StringSerde(), new LongSerde()))
           .withReadFunction(readFn)
-          .withRateLimiter(readRateLimiter, null, null)
-          .withSerde(KVSerde.of(new StringSerde(), new LongSerde())));
-      tableDescriptors.add(new RocksDbTableDescriptor("local-table-1")
-          .withBlockSize(4096)
-          .withSerde(KVSerde.of(new StringSerde(), new StringSerde())));
+          .withRateLimiter(readRateLimiter, null, null));
+      tableDescriptors.add(new RocksDbTableDescriptor("local-table-1", KVSerde.of(new StringSerde(), new StringSerde()))
+          .withBlockSize(4096));
       return tableDescriptors;
     }
   }
