From b75d744890d06b83bf72aa0246d8ac90feb3062e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=99=BD=E9=B5=BA?= Date: Sun, 9 Aug 2026 00:15:46 +0800 Subject: [PATCH] [client] Scope sender failures to per-table requests --- .../org/apache/fluss/client/write/Sender.java | 37 ++-- .../apache/fluss/client/write/SenderTest.java | 190 +++++++++++++++++- .../tablet/TestTabletServerGateway.java | 22 +- 3 files changed, 212 insertions(+), 37 deletions(-) diff --git a/fluss-client/src/main/java/org/apache/fluss/client/write/Sender.java b/fluss-client/src/main/java/org/apache/fluss/client/write/Sender.java index f1648823ad5..b16780035f7 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/write/Sender.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/write/Sender.java @@ -376,24 +376,20 @@ private void sendWriteRequest(int destination, short acks, List } // group record batch by table id. - final Map recordsByBucket = new HashMap<>(); Map> writeBatchByTable = new HashMap<>(); batches.forEach( - batch -> { - // keep the batch before ack. - recordsByBucket.put(batch.tableBucket(), batch); - writeBatchByTable - .computeIfAbsent( - batch.tableBucket().getTableId(), k -> new ArrayList<>()) - .add(batch); - }); + batch -> + writeBatchByTable + .computeIfAbsent( + batch.tableBucket().getTableId(), k -> new ArrayList<>()) + .add(batch)); TabletServerGateway gateway = metadataUpdater.newTabletServerClientForNode(destination); if (gateway == null) { handleWriteRequestException( new LeaderNotAvailableException( "Server " + destination + " is not found in metadata cache."), - recordsByBucket); + batches); } else { writeBatchByTable.forEach( (tableId, writeBatches) -> { @@ -403,14 +399,14 @@ private void sendWriteRequest(int destination, short acks, List makeProduceLogRequest( tableId, acks, maxRequestTimeoutMs, writeBatches), tableId, - recordsByBucket); + writeBatches); } else { sendPutKvRequestAndHandleResponse( gateway, makePutKvRequest( tableId, acks, maxRequestTimeoutMs, writeBatches), tableId, - recordsByBucket); + writeBatches); } }); } @@ -433,7 +429,9 @@ private void sendProduceLogRequestAndHandleResponse( TabletServerGateway gateway, ProduceLogRequest request, long tableId, - Map recordsByBucket) { + List writeBatches) { + Map recordsByBucket = new HashMap<>(); + writeBatches.forEach(batch -> recordsByBucket.put(batch.tableBucket(), batch)); long startTime = System.currentTimeMillis(); gateway.produceLog(request) .whenComplete( @@ -441,7 +439,7 @@ private void sendProduceLogRequestAndHandleResponse( writerMetricGroup.setSendLatencyInMs( System.currentTimeMillis() - startTime); if (e != null) { - handleWriteRequestException(e, recordsByBucket); + handleWriteRequestException(e, writeBatches); } else { handleProduceLogResponse( produceLogResponse, tableId, recordsByBucket); @@ -453,7 +451,9 @@ private void sendPutKvRequestAndHandleResponse( TabletServerGateway gateway, PutKvRequest request, long tableId, - Map recordsByBucket) { + List writeBatches) { + Map recordsByBucket = new HashMap<>(); + writeBatches.forEach(batch -> recordsByBucket.put(batch.tableBucket(), batch)); long startTime = System.currentTimeMillis(); gateway.putKv(request) .whenComplete( @@ -461,7 +461,7 @@ private void sendPutKvRequestAndHandleResponse( writerMetricGroup.setSendLatencyInMs( System.currentTimeMillis() - startTime); if (e != null) { - handleWriteRequestException(e, recordsByBucket); + handleWriteRequestException(e, writeBatches); } else { handlePutKvResponse(putKvResponse, tableId, recordsByBucket); } @@ -530,14 +530,13 @@ private void handlePutKvResponse( metadataUpdater.invalidPhysicalTableBucketMeta(invalidMetadataTablesSet); } - private void handleWriteRequestException( - Throwable t, Map recordsByBucket) { + private void handleWriteRequestException(Throwable t, List writeBatches) { ApiError error = ApiError.fromThrowable(t); // if batch failed because of retrievable exception, we need to retry send all those // batches. Set invalidMetadataTablesSet = new HashSet<>(); - for (ReadyWriteBatch batch : recordsByBucket.values()) { + for (ReadyWriteBatch batch : writeBatches) { Set invalidMetadataTables = handleWriteBatchException(batch, error); invalidMetadataTablesSet.addAll(invalidMetadataTables); } diff --git a/fluss-client/src/test/java/org/apache/fluss/client/write/SenderTest.java b/fluss-client/src/test/java/org/apache/fluss/client/write/SenderTest.java index b222c9e3c18..69d9c669c9e 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/write/SenderTest.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/write/SenderTest.java @@ -24,6 +24,8 @@ import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; import org.apache.fluss.config.MemorySize; +import org.apache.fluss.exception.AuthorizationException; +import org.apache.fluss.exception.NetworkException; import org.apache.fluss.exception.TableNotExistException; import org.apache.fluss.exception.TimeoutException; import org.apache.fluss.metadata.PhysicalTablePath; @@ -39,6 +41,7 @@ import org.apache.fluss.rpc.messages.ApiMessage; import org.apache.fluss.rpc.messages.ProduceLogRequest; import org.apache.fluss.rpc.messages.ProduceLogResponse; +import org.apache.fluss.rpc.messages.PutKvRequest; import org.apache.fluss.rpc.messages.PutKvResponse; import org.apache.fluss.rpc.protocol.Errors; import org.apache.fluss.server.tablet.TestTabletServerGateway; @@ -60,10 +63,10 @@ import java.util.concurrent.CompletableFuture; import static org.apache.fluss.record.LogRecordBatchFormat.NO_WRITER_ID; -import static org.apache.fluss.record.TestData.DATA1_PHYSICAL_TABLE_PATH; import static org.apache.fluss.record.TestData.DATA1_ROW_TYPE; import static org.apache.fluss.record.TestData.DATA1_SCHEMA_PK; import static org.apache.fluss.record.TestData.DATA1_TABLE_DESCRIPTOR; +import static org.apache.fluss.record.TestData.DATA1_TABLE_DESCRIPTOR_PK; import static org.apache.fluss.record.TestData.DATA1_TABLE_ID; import static org.apache.fluss.record.TestData.DATA1_TABLE_ID_PK; import static org.apache.fluss.record.TestData.DATA1_TABLE_INFO; @@ -508,12 +511,14 @@ void testRetryAfterResettingInFlightBatchSequence() throws Exception { assertThat(idempotenceManager.hasInflightBatches(tb1)).isFalse(); assertThat( accumulator.getReadyDeque( - DATA1_PHYSICAL_TABLE_PATH, tb1.getBucket())) + PhysicalTablePath.of(DATA1_TABLE_PATH), + tb1.getBucket())) .hasSize(1); assertThat( accumulator .getReadyDeque( - DATA1_PHYSICAL_TABLE_PATH, tb1.getBucket()) + PhysicalTablePath.of(DATA1_TABLE_PATH), + tb1.getBucket()) .peek() .batchSequence()) .isEqualTo(0); @@ -559,7 +564,7 @@ void testCorrectHandlingOfOutOfOrderResponses() throws Exception { sender1.runOnce(); // receive response 1. Deque queuedBatches = - accumulator.getReadyDeque(DATA1_PHYSICAL_TABLE_PATH, tb1.getBucket()); + accumulator.getReadyDeque(PhysicalTablePath.of(DATA1_TABLE_PATH), tb1.getBucket()); // Make sure that we are queueing the second batch first. assertThat(queuedBatches.size()).isEqualTo(1); @@ -629,7 +634,7 @@ void testCorrectHandlingOfOutOfOrderResponsesWhenSecondSucceeds() throws Excepti assertThat(future2.get()).isNull(); assertThat(future1.isDone()).isFalse(); Deque queuedBatches = - accumulator.getReadyDeque(DATA1_PHYSICAL_TABLE_PATH, tb1.getBucket()); + accumulator.getReadyDeque(PhysicalTablePath.of(DATA1_TABLE_PATH), tb1.getBucket()); assertThat(queuedBatches.size()).isEqualTo(0); assertThat(idempotenceManager.lastAckedBatchSequence(tb1)).isEqualTo(Optional.of(1)); @@ -851,6 +856,52 @@ void testSendWhenDestinationIsNullInMetadata() throws Exception { assertThat(future.get()).isNull(); } + @Test + void testProduceLogRpcFailureOnlyFailsOwnedTableBatches() throws Exception { + TablePath secondTablePath = TablePath.of("test_db_2", "test_log_table_2"); + TableInfo secondTableInfo = + TableInfo.of( + secondTablePath, + DATA2_TABLE_ID, + 1, + DATA1_TABLE_DESCRIPTOR, + DEFAULT_REMOTE_DATA_DIR, + System.currentTimeMillis(), + System.currentTimeMillis()); + resetTableInfosWith(secondTableInfo); + + TableBucket secondTableBucket = new TableBucket(DATA2_TABLE_ID, 0); + CompletableFuture firstFuture = new CompletableFuture<>(); + CompletableFuture secondFuture = new CompletableFuture<>(); + appendToAccumulator(tb1, row(1, "a"), (tb, leo, e) -> firstFuture.complete(e)); + appendToAccumulator( + secondTableInfo, + secondTableBucket, + row(2, "b"), + (tb, leo, e) -> secondFuture.complete(e)); + + sender.runOnce(); + assertThat(pendingWriteRequestTableIds(tb1)) + .containsExactlyInAnyOrder(DATA1_TABLE_ID, DATA2_TABLE_ID); + + failRequest( + tb1, + findRequestIndex(tb1, DATA1_TABLE_ID), + new AuthorizationException("first table is not authorized")); + + assertThat(firstFuture.get()).isInstanceOf(AuthorizationException.class); + assertThat(secondFuture.isDone()).isFalse(); + assertThat(sender.numOfInFlightBatches(secondTableBucket)).isEqualTo(1); + assertThat(pendingWriteRequestTableIds(tb1)).containsExactly(DATA2_TABLE_ID); + + finishRequest( + tb1, + findRequestIndex(tb1, DATA2_TABLE_ID), + createProduceLogResponse(secondTableBucket, 0L, 1L)); + assertThat(secondFuture.get()).isNull(); + assertThat(sender.numOfInFlightBatches(secondTableBucket)).isEqualTo(0); + } + @Test void testRetryPutKeyWithSchemaNotExistException() throws Exception { TableBucket tableBucket = new TableBucket(DATA1_TABLE_ID_PK, 0); @@ -957,6 +1008,73 @@ void testPutKvStorageExceptionResponseRetriesInsteadOfFailing() throws Exception assertThat(future.get()).isNull(); } + @Test + void testPutKvRpcFailuresRetryOnlyOwnedTableBatches() throws Exception { + TablePath secondTablePath = TablePath.of("test_db_2", "test_pk_table_2"); + TableInfo secondTableInfo = + TableInfo.of( + secondTablePath, + DATA2_TABLE_ID, + 1, + DATA1_TABLE_DESCRIPTOR_PK, + DEFAULT_REMOTE_DATA_DIR, + System.currentTimeMillis(), + System.currentTimeMillis()); + resetTableInfosWith(secondTableInfo); + + TableBucket firstTableBucket = new TableBucket(DATA1_TABLE_ID_PK, 0); + TableBucket secondTableBucket = new TableBucket(DATA2_TABLE_ID, 0); + CompletableFuture firstFuture = new CompletableFuture<>(); + CompletableFuture secondFuture = new CompletableFuture<>(); + appendKvToAccumulator( + firstTableBucket, + compactedRow(DATA1_ROW_TYPE, new Object[] {1, "a"}), + (tb, leo, e) -> firstFuture.complete(e)); + appendKvToAccumulator( + secondTableInfo, + secondTableBucket, + compactedRow(DATA1_ROW_TYPE, new Object[] {2, "b"}), + (tb, leo, e) -> secondFuture.complete(e)); + + sender.runOnce(); + assertThat(pendingWriteRequestTableIds(tb1)) + .containsExactlyInAnyOrder(DATA1_TABLE_ID_PK, DATA2_TABLE_ID); + Cluster clusterBeforeFailures = metadataUpdater.getCluster(); + + failRequest( + tb1, + findRequestIndex(tb1, DATA1_TABLE_ID_PK), + new NetworkException("first table request failed")); + assertThat(writerMetricGroup.recordsRetryTotal().getCount()).isEqualTo(1L); + assertThat(sender.numOfInFlightBatches(firstTableBucket)).isEqualTo(0); + assertThat(sender.numOfInFlightBatches(secondTableBucket)).isEqualTo(1); + + failRequest( + tb1, + findRequestIndex(tb1, DATA2_TABLE_ID), + new NetworkException("second table request failed")); + assertThat(firstFuture.isDone()).isFalse(); + assertThat(secondFuture.isDone()).isFalse(); + assertThat(writerMetricGroup.recordsRetryTotal().getCount()).isEqualTo(2L); + assertThat(sender.numOfInFlightBatches(secondTableBucket)).isEqualTo(0); + + metadataUpdater.updateCluster(clusterBeforeFailures); + sender.runOnce(); + assertThat(pendingWriteRequestTableIds(tb1)) + .containsExactlyInAnyOrder(DATA1_TABLE_ID_PK, DATA2_TABLE_ID); + + finishRequest( + tb1, + findRequestIndex(tb1, DATA1_TABLE_ID_PK), + createPutKvResponse(firstTableBucket, 1L)); + finishRequest( + tb1, + findRequestIndex(tb1, DATA2_TABLE_ID), + createPutKvResponse(secondTableBucket, 1L)); + assertThat(firstFuture.get()).isNull(); + assertThat(secondFuture.get()).isNull(); + } + @Test void testSendWhenTableIdChanges() throws Exception { CompletableFuture future1 = new CompletableFuture<>(); @@ -998,6 +1116,14 @@ private TestingMetadataUpdater initializeMetadataUpdater() { return new TestingMetadataUpdater(tableInfos); } + private void resetTableInfosWith(TableInfo tableInfo) { + Map tableInfos = new HashMap<>(); + tableInfos.put(DATA1_TABLE_PATH, DATA1_TABLE_INFO); + tableInfos.put(DATA1_TABLE_PATH_PK, DATA1_TABLE_INFO_PK); + tableInfos.put(tableInfo.getTablePath(), tableInfo); + metadataUpdater.updateTableInfos(tableInfos); + } + private void appendToAccumulator(TableBucket tb, GenericRow row, WriteCallback writeCallback) throws Exception { appendToAccumulator(DATA1_TABLE_INFO, tb, row, writeCallback); @@ -1007,7 +1133,8 @@ private void appendToAccumulator( TableInfo tableInfo, TableBucket tb, GenericRow row, WriteCallback writeCallback) throws Exception { accumulator.append( - WriteRecord.forArrowAppend(tableInfo, DATA1_PHYSICAL_TABLE_PATH, row, null), + WriteRecord.forArrowAppend( + tableInfo, PhysicalTablePath.of(tableInfo.getTablePath()), row, null), writeCallback, metadataUpdater.getCluster(), tb.getBucket(), @@ -1016,12 +1143,21 @@ private void appendToAccumulator( private void appendKvToAccumulator( TableBucket tableBucket, BinaryRow row, WriteCallback writeCallback) throws Exception { + appendKvToAccumulator(DATA1_TABLE_INFO_PK, tableBucket, row, writeCallback); + } + + private void appendKvToAccumulator( + TableInfo tableInfo, + TableBucket tableBucket, + BinaryRow row, + WriteCallback writeCallback) + throws Exception { int[] pkIndex = DATA1_SCHEMA_PK.getPrimaryKeyIndexes(); byte[] key = new CompactedKeyEncoder(DATA1_ROW_TYPE, pkIndex).encodeKey(row); accumulator.append( WriteRecord.forUpsert( - DATA1_TABLE_INFO_PK, - PhysicalTablePath.of(DATA1_TABLE_PATH_PK), + tableInfo, + PhysicalTablePath.of(tableInfo.getTablePath()), row, key, key, @@ -1057,6 +1193,44 @@ private int pendingRequestSize(TableBucket tb) { return gateway.pendingRequestSize(); } + private int findRequestIndex(TableBucket referenceBucket, long tableId) { + int requestCount = pendingRequestSize(referenceBucket); + for (int index = 0; index < requestCount; index++) { + if (getWriteRequestTableId(getRequest(referenceBucket, index)) == tableId) { + return index; + } + } + throw new IllegalStateException("No pending write request for table " + tableId); + } + + private List pendingWriteRequestTableIds(TableBucket referenceBucket) { + List tableIds = new ArrayList<>(); + int requestCount = pendingRequestSize(referenceBucket); + for (int index = 0; index < requestCount; index++) { + tableIds.add(getWriteRequestTableId(getRequest(referenceBucket, index))); + } + return tableIds; + } + + private static long getWriteRequestTableId(ApiMessage request) { + if (request instanceof ProduceLogRequest) { + return ((ProduceLogRequest) request).getTableId(); + } + if (request instanceof PutKvRequest) { + return ((PutKvRequest) request).getTableId(); + } + throw new IllegalArgumentException( + "Expected a write request but found " + request.getClass().getName()); + } + + private void failRequest(TableBucket referenceBucket, int index, Throwable throwable) { + TestTabletServerGateway gateway = + (TestTabletServerGateway) + metadataUpdater.newTabletServerClientForNode( + metadataUpdater.leaderFor(DATA1_TABLE_PATH, referenceBucket)); + gateway.failRequest(index, throwable); + } + private void finishIdempotentProduceLogRequest( int batchSequence, TableBucket tb, int index, ProduceLogResponse response) { TestTabletServerGateway gateway = diff --git a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TestTabletServerGateway.java b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TestTabletServerGateway.java index c78270ea5ca..06ff7078e20 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TestTabletServerGateway.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TestTabletServerGateway.java @@ -404,35 +404,37 @@ public ApiMessage getRequest(int index) { } public void response(int index, ApiMessage response) { + removeRequestFuture(index).complete(response); + } + + /** Completes the request at the given index exceptionally. */ + public void failRequest(int index, Throwable throwable) { + removeRequestFuture(index).completeExceptionally(throwable); + } + + @SuppressWarnings("unchecked") + private CompletableFuture removeRequestFuture(int index) { if (requests.isEmpty()) { throw new IllegalStateException("No requests pending for inbound response."); } - // Index out of bounds check. if (index >= requests.size()) { throw new IllegalArgumentException( "Index " + index + " is out of bounds for requests queue."); } - CompletableFuture result = null; int currentIndex = 0; for (Iterator>> it = requests.iterator(); it.hasNext(); ) { Tuple2> tuple = it.next(); if (currentIndex == index) { - result = (CompletableFuture) tuple.f1; it.remove(); - break; + return (CompletableFuture) tuple.f1; } currentIndex++; } - if (result != null) { - result.complete(response); - } else { - throw new IllegalStateException( - "The future to complete was not found at index " + index); - } + throw new IllegalStateException("The future to complete was not found at index " + index); } private StopReplicaResponse mockStopReplicaResponse(