SAMZA-2026: Refactor remote table API to separate retry policy settings

As per subject, the goal is to make configuration of retry policies consistent with other API's.

Author: Wei Song <[email protected]>

Reviewers: Aditya Toomula <[email protected]>

Closes apache#842 from weisong44/SAMZA-2026

Author: weisong44

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

@@ -125,29 +125,25 @@ public RemoteTableDescriptor<K, V> withWriteFunction(TableWriteFunction<K, V> wr
   }
 
   /**
-   * Use specified TableReadFunction with remote table.
-   * @param readFn read function instance
-   * @param retryPolicy retry policy for the read function
+   * Use specified {@link TableRetryPolicy} with the {@link TableReadFunction}.
+   * @param retryPolicy retry policy for the write function
    * @return this table descriptor instance
    */
-  public RemoteTableDescriptor<K, V> withReadFunction(TableReadFunction<K, V> readFn, TableRetryPolicy retryPolicy) {
+  public RemoteTableDescriptor<K, V> withReadRetryPolicy(TableRetryPolicy retryPolicy) {
     Preconditions.checkNotNull(readFn, "null read function");
     Preconditions.checkNotNull(retryPolicy, "null retry policy");
-    this.readFn = readFn;
     this.readRetryPolicy = retryPolicy;
     return this;
   }
 
   /**
-   * Use specified TableWriteFunction with remote table.
-   * @param writeFn write function instance
+   * Use specified {@link TableRetryPolicy} with the {@link TableWriteFunction}.
    * @param retryPolicy retry policy for the write function
    * @return this table descriptor instance
    */
-  public RemoteTableDescriptor<K, V> withWriteFunction(TableWriteFunction<K, V> writeFn, TableRetryPolicy retryPolicy) {
+  public RemoteTableDescriptor<K, V> withWriteRetryPolicy(TableRetryPolicy retryPolicy) {
     Preconditions.checkNotNull(writeFn, "null write function");
     Preconditions.checkNotNull(retryPolicy, "null retry policy");
-    this.writeFn = writeFn;
     this.writeRetryPolicy = retryPolicy;
     return this;
   }

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

@@ -175,12 +175,11 @@ public int getCredits(K key, V value) {
 
   private void doTestDeserializeReadFunctionAndLimiter(boolean rateOnly, boolean rlGets, boolean rlPuts) {
     int numRateLimitOps = (rlGets ? 1 : 0) + (rlPuts ? 1 : 0);
-    RemoteTableDescriptor<String, String> desc = new RemoteTableDescriptor("1");
-    TableRetryPolicy retryPolicy = new TableRetryPolicy();
-    retryPolicy.withRetryPredicate((ex) -> false);
-    desc.withReadFunction(createMockTableReadFunction(), retryPolicy);
-    desc.withWriteFunction(createMockTableWriteFunction());
-    desc.withAsyncCallbackExecutorPoolSize(10);
+    RemoteTableDescriptor<String, String> desc = new RemoteTableDescriptor("1")
+        .withReadFunction(createMockTableReadFunction())
+        .withReadRetryPolicy(new TableRetryPolicy().withRetryPredicate((ex) -> false))
+        .withWriteFunction(createMockTableWriteFunction())
+        .withAsyncCallbackExecutorPoolSize(10);
 
     if (rateOnly) {
       if (rlGets) {
@@ -218,8 +217,8 @@ private void doTestDeserializeReadFunctionAndLimiter(boolean rateOnly, boolean r
     ThreadPoolExecutor callbackExecutor = (ThreadPoolExecutor) rwTable.getCallbackExecutor();
     Assert.assertEquals(10, callbackExecutor.getCorePoolSize());
 
-    Assert.assertNotNull(rwTable.getReadFn() instanceof RetriableReadFunction);
-    Assert.assertNotNull(!(rwTable.getWriteFn() instanceof RetriableWriteFunction));
+    Assert.assertTrue(rwTable.getReadFn() instanceof RetriableReadFunction);
+    Assert.assertFalse(rwTable.getWriteFn() instanceof RetriableWriteFunction);
   }
 
   @Test