[client] Fix infinite retry loop in idempotent writer on lost response with OutOfOrderSequenceException (apache#2827)

Author: LiebingYu

fluss-client/src/main/java/org/apache/fluss/client/write/IdempotenceManager.java

@@ -294,6 +294,23 @@ synchronized boolean canRetry(WriteBatch batch, TableBucket tableBucket, Errors
         return false;
     }
 
+    /**
+     * Returns true if the batch has already been committed on the server.
+     *
+     * <p>If the batch's sequence is less than or equal to {@code lastAckedBatchSequence}, it means
+     * a higher-sequence batch has already been acknowledged. This implies the previous batch was
+     * also successfully written on the server (otherwise the higher-sequence batch could not have
+     * been committed).
+     *
+     * @param batch the batch to check
+     * @param tableBucket the target table-bucket
+     * @return true if the batch is already committed on the server
+     */
+    synchronized boolean isAlreadyCommitted(WriteBatch batch, TableBucket tableBucket) {
+        Optional<Integer> lastAcked = lastAckedBatchSequence(tableBucket);
+        return lastAcked.isPresent() && batch.batchSequence() <= lastAcked.get();
+    }
+
     void maybeWaitForWriterId(Set<PhysicalTablePath> tablePaths) throws Throwable {
         if (isWriterIdValid()) {
             return;

fluss-client/src/main/java/org/apache/fluss/client/write/Sender.java

@@ -532,7 +532,27 @@ private Set<PhysicalTablePath> handleWriteBatchException(
             ReadyWriteBatch readyWriteBatch, ApiError error) {
         Set<PhysicalTablePath> invalidMetadataTables = new HashSet<>();
         WriteBatch writeBatch = readyWriteBatch.writeBatch();
-        if (canRetry(readyWriteBatch, error.error())) {
+        if (error.error() == Errors.DUPLICATE_SEQUENCE_EXCEPTION) {
+            // If we have received a duplicate batch sequence error, it means that the batch
+            // sequence has advanced beyond the sequence of the current batch.
+            // The only thing we can do is to return success to the user.
+            completeBatch(readyWriteBatch);
+        } else if (error.error() == Errors.OUT_OF_ORDER_SEQUENCE_EXCEPTION
+                && idempotenceManager.idempotenceEnabled()
+                && idempotenceManager.isAlreadyCommitted(
+                        writeBatch, readyWriteBatch.tableBucket())) {
+            // The batch received OUT_OF_ORDER_SEQUENCE_EXCEPTION but its sequence is already
+            // <= lastAckedBatchSequence, which means it was successfully written on the server
+            // but the response was lost. Complete it as success to avoid infinite retry loop.
+            LOG.warn(
+                    "Batch for table-bucket {} with sequence {} received "
+                            + "OUT_OF_ORDER_SEQUENCE_EXCEPTION but has already been committed "
+                            + "(lastAckedBatchSequence={}). Treating as success due to lost response.",
+                    readyWriteBatch.tableBucket(),
+                    writeBatch.batchSequence(),
+                    idempotenceManager.lastAckedBatchSequence(readyWriteBatch.tableBucket()));
+            completeBatch(readyWriteBatch);
+        } else if (canRetry(readyWriteBatch, error.error())) {
             // if batch failed because of retrievable exception, we need to retry send all those
             // batches.
             LOG.warn(
@@ -575,11 +595,6 @@ private Set<PhysicalTablePath> handleWriteBatchException(
                 }
                 invalidMetadataTables.add(writeBatch.physicalTablePath());
             }
-        } else if (error.error() == Errors.DUPLICATE_SEQUENCE_EXCEPTION) {
-            // If we have received a duplicate batch sequence error, it means that the batch
-            // sequence has advanced beyond the sequence of the current batch.
-            // The only thing we can do is to return success to the user.
-            completeBatch(readyWriteBatch);
         } else {
             LOG.warn(
                     "Get error write response on table bucket {}, fail. Error: {}",

fluss-client/src/test/java/org/apache/fluss/client/write/SenderTest.java

@@ -610,6 +610,100 @@ void testCorrectHandlingOfOutOfOrderResponsesWhenSecondSucceeds() throws Excepti
         assertThat(future1.get()).isNull();
     }
 
+    /**
+     * Tests that when a batch's response is lost (e.g., due to request timeout) but the batch was
+     * successfully written on the server, and subsequent batches with higher sequence numbers are
+     * acknowledged, the client should treat the retried batch as already committed instead of
+     * entering an infinite retry loop with {@link
+     * org.apache.fluss.exception.OutOfOrderSequenceException}.
+     *
+     * <p>Detailed scenario:
+     *
+     * <ol>
+     *   <li>Send batch1(seq=0) ~ batch5(seq=4). All 5 batches are successfully written on the
+     *       server (server {@code lastBatchSeq=4}).
+     *   <li>batch2~5 (seq=1~4) responses return normally. Client {@code lastAckedBatchSequence=4}.
+     *   <li>Send batch6(seq=5) and ack successfully. Server {@code lastBatchSeq=5}.
+     *   <li>batch1(seq=0) response is lost due to {@code REQUEST_TIME_OUT}. batch1 is re-enqueued
+     *       for retry.
+     *   <li>Client retries batch1(seq=0). Since server {@code lastBatchSeq=5} and {@code 0 != 5+1},
+     *       server returns {@code OUT_OF_ORDER_SEQUENCE_EXCEPTION}.
+     *   <li>Client detects {@code batch1.seq(0) <= lastAckedBatchSequence(5)}: batch1 is already
+     *       committed. Client completes batch1 successfully without further retries.
+     * </ol>
+     */
+    @Test
+    void testCorrectHandlingOfOutOfOrderResponsesWhenResponseLostButSubsequentBatchesSucceeded()
+            throws Exception {
+        IdempotenceManager idempotenceManager = createIdempotenceManager(true);
+        Sender sender1 = setupWithIdempotenceState(idempotenceManager);
+        sender1.runOnce();
+        assertThat(idempotenceManager.isWriterIdValid()).isTrue();
+        assertThat(idempotenceManager.nextSequence(tb1)).isEqualTo(0);
+
+        // Send batch1 (seq=0): its response will be lost later.
+        CompletableFuture<Exception> future1 = new CompletableFuture<>();
+        appendToAccumulator(tb1, row(1, "a"), (tb, leo, e) -> future1.complete(e));
+        sender1.runOnce();
+        assertThat(future1.isDone()).isFalse();
+        assertThat(idempotenceManager.nextSequence(tb1)).isEqualTo(1);
+        assertThat(idempotenceManager.lastAckedBatchSequence(tb1)).isNotPresent();
+
+        // Send batch2~5 (seq=1~4) and collect their futures.
+        int numFollowingBatches = 4;
+        List<CompletableFuture<Exception>> followingFutures = new ArrayList<>();
+        for (int i = 0; i < numFollowingBatches; i++) {
+            CompletableFuture<Exception> future = new CompletableFuture<>();
+            followingFutures.add(future);
+            appendToAccumulator(tb1, row(i + 2, "b"), (tb, leo, e) -> future.complete(e));
+            sender1.runOnce();
+            assertThat(future.isDone()).isFalse();
+        }
+        assertThat(idempotenceManager.nextSequence(tb1)).isEqualTo(5);
+
+        // batch2~5 (seq=1~4) responses return normally.
+        for (int seq = 1; seq <= numFollowingBatches; seq++) {
+            finishIdempotentProduceLogRequest(
+                    seq, tb1, 1, createProduceLogResponse(tb1, seq, seq + 1L));
+            assertThat(idempotenceManager.lastAckedBatchSequence(tb1)).isEqualTo(Optional.of(seq));
+            assertThat(followingFutures.get(seq - 1).isDone()).isTrue();
+            assertThat(followingFutures.get(seq - 1).get()).isNull();
+        }
+
+        // Send batch6 (seq=5) and ack successfully.
+        // Now server lastBatchSeq=5. batch1 (seq=0) is still waiting response.
+        CompletableFuture<Exception> future6 = new CompletableFuture<>();
+        appendToAccumulator(tb1, row(6, "f"), (tb, leo, e) -> future6.complete(e));
+        sender1.runOnce(); // drain and send batch6 (seq=5)
+        finishIdempotentProduceLogRequest(5, tb1, 1, createProduceLogResponse(tb1, 5L, 6L));
+        assertThat(idempotenceManager.lastAckedBatchSequence(tb1)).isEqualTo(Optional.of(5));
+        assertThat(future6.isDone()).isTrue();
+        assertThat(future6.get()).isNull();
+
+        // All 6 batches are written successfully on the server (server lastBatchSeq=5).
+        // batch1 (seq=0) response is lost, simulated by REQUEST_TIME_OUT.
+        finishIdempotentProduceLogRequest(
+                0, tb1, 0, createProduceLogResponse(tb1, Errors.REQUEST_TIME_OUT));
+        assertThat(future1.isDone()).isFalse();
+
+        // Now retry batch1 (seq=0). Server lastBatchSeq=5, so 0 != 5+1,
+        // server returns OUT_OF_ORDER_SEQUENCE_EXCEPTION.
+        sender1.runOnce(); // send retried batch1
+        finishIdempotentProduceLogRequest(
+                0, tb1, 0, createProduceLogResponse(tb1, Errors.OUT_OF_ORDER_SEQUENCE_EXCEPTION));
+
+        // The client should detect that batch1.seq(0) <= lastAckedBatchSequence(5),
+        // meaning batch1 was already committed on the server (its response was just lost).
+        // It should complete batch1 successfully instead of entering an infinite retry loop.
+        assertThat(future1.isDone()).isTrue();
+        assertThat(future1.get()).isNull();
+        // lastAckedBatchSequence should remain at 5 (not changed by completing already-committed
+        // batch1)
+        assertThat(idempotenceManager.lastAckedBatchSequence(tb1)).isEqualTo(Optional.of(5));
+        // No more inflight batches
+        assertThat(sender1.numOfInFlightBatches(tb1)).isEqualTo(0);
+    }
+
     @Test
     void testCorrectHandlingOfDuplicateSequenceError() throws Exception {
         IdempotenceManager idempotenceManager = createIdempotenceManager(true);