SAMZA-1964: Make getTableSpec() in RemoteTableDescriptor reentrant

As per subject

Author: Wei Song <[email protected]>

Reviewers: Xinyu Liu <[email protected]>

Closes apache#747 from weisong44/SAMZA-1964

Author: weisong44

samza-core/src/main/java/org/apache/samza/table/remote/descriptors/RemoteTableDescriptor.java

@@ -65,8 +65,7 @@ public class RemoteTableDescriptor<K, V> extends BaseTableDescriptor<K, V, Remot
   // Output support for a specific remote store (optional)
   private TableWriteFunction<K, V> writeFn;
 
-  // Rate limiter for client-side throttling;
-  // can either be constructed indirectly from rates or overridden by withRateLimiter()
+  // Rate limiter for client-side throttling; it is set by withRateLimiter()
   private RateLimiter rateLimiter;
 
   // Rates for constructing the default rate limiter when they are non-zero
@@ -113,21 +112,19 @@ public TableSpec getTableSpec() {
       tableSpecConfig.put(RemoteTableProvider.WRITE_FN, SerdeUtils.serialize("write function", writeFn));
     }
 
-    // Serialize the rate limiter if specified
     if (!tagCreditsMap.isEmpty()) {
-      rateLimiter = new EmbeddedTaggedRateLimiter(tagCreditsMap);
-    }
-
-    if (rateLimiter != null) {
+      tableSpecConfig.put(RemoteTableProvider.RATE_LIMITER, SerdeUtils.serialize("rate limiter",
+          new EmbeddedTaggedRateLimiter(tagCreditsMap)));
+    } else if (rateLimiter != null) {
       tableSpecConfig.put(RemoteTableProvider.RATE_LIMITER, SerdeUtils.serialize("rate limiter", rateLimiter));
     }
 
-    // Serialize the readCredit and writeCredit functions
+    // Serialize the readCredit functions
     if (readCreditFn != null) {
       tableSpecConfig.put(RemoteTableProvider.READ_CREDIT_FN, SerdeUtils.serialize(
           "read credit function", readCreditFn));
     }
-
+    // Serialize the writeCredit functions
     if (writeCreditFn != null) {
       tableSpecConfig.put(RemoteTableProvider.WRITE_CREDIT_FN, SerdeUtils.serialize(
           "write credit function", writeCreditFn));

samza-core/src/test/java/org/apache/samza/table/remote/descriptors/TestRemoteTableDescriptor.java

@@ -197,6 +197,7 @@ private void doTestDeserializeReadFunctionAndLimiter(boolean rateOnly, boolean r
     }
 
     TableSpec spec = desc.getTableSpec();
+    spec = desc.getTableSpec();
     RemoteTableProvider provider = new RemoteTableProvider(spec);
     provider.init(createMockContext());
     Table table = provider.getTable();

samza-kv-inmemory/src/test/java/org/apache/samza/storage/kv/inmemory/descriptors/TestInMemoryTableDescriptor.java

@@ -28,6 +28,13 @@
 
 
 public class TestInMemoryTableDescriptor {
+  @Test
+  public void testMinimal() {
+    InMemoryTableDescriptor tableDescriptor = new InMemoryTableDescriptor("1");
+    tableDescriptor.getTableSpec();
+    tableDescriptor.getTableSpec();
+  }
+
   @Test
   public void testTableSpec() {
 

samza-kv-rocksdb/src/test/java/org/apache/samza/storage/kv/descriptors/TestRocksDbTableDescriptor.java

@@ -30,8 +30,10 @@ public class TestRocksDbTableDescriptor {
 
   @Test
   public void testMinimal() {
-    new RocksDbTableDescriptor<Integer, String>("1")
-        .validate();
+    RocksDbTableDescriptor tableDescriptor = new RocksDbTableDescriptor("1");
+    tableDescriptor.validate();
+    tableDescriptor.getTableSpec();
+    tableDescriptor.getTableSpec();
   }
 
   @Test