Skip to content

[client] Scope sender failures to per-table requests #4017

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
platinumhamburg wants to merge 1 commit into
base: main
Choose a base branch
from
+212 −37
Open

[client] Scope sender failures to per-table requests #4017

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 18 additions & 19 deletions fluss-client/src/main/java/org/apache/fluss/client/write/Sender.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -376,24 +376,20 @@ private void sendWriteRequest(int destination, short acks, List<ReadyWriteBatch>
}

// group record batch by table id.
final Map<TableBucket, ReadyWriteBatch> recordsByBucket = new HashMap<>();
Map<Long, List<ReadyWriteBatch>> 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) -> {
Expand All @@ -403,14 +399,14 @@ private void sendWriteRequest(int destination, short acks, List<ReadyWriteBatch>
makeProduceLogRequest(
tableId, acks, maxRequestTimeoutMs, writeBatches),
tableId,
recordsByBucket);
writeBatches);
} else {
sendPutKvRequestAndHandleResponse(
gateway,
makePutKvRequest(
tableId, acks, maxRequestTimeoutMs, writeBatches),
tableId,
recordsByBucket);
writeBatches);
}
});
}
Expand All @@ -433,15 +429,17 @@ private void sendProduceLogRequestAndHandleResponse(
TabletServerGateway gateway,
ProduceLogRequest request,
long tableId,
Map<TableBucket, ReadyWriteBatch> recordsByBucket) {
List<ReadyWriteBatch> writeBatches) {
Map<TableBucket, ReadyWriteBatch> recordsByBucket = new HashMap<>();
writeBatches.forEach(batch -> recordsByBucket.put(batch.tableBucket(), batch));
Comment thread
platinumhamburg marked this conversation as resolved.
long startTime = System.currentTimeMillis();
gateway.produceLog(request)
.whenComplete(
(produceLogResponse, e) -> {
writerMetricGroup.setSendLatencyInMs(
System.currentTimeMillis() - startTime);
if (e != null) {
handleWriteRequestException(e, recordsByBucket);
handleWriteRequestException(e, writeBatches);
} else {
handleProduceLogResponse(
produceLogResponse, tableId, recordsByBucket);
Expand All @@ -453,15 +451,17 @@ private void sendPutKvRequestAndHandleResponse(
TabletServerGateway gateway,
PutKvRequest request,
long tableId,
Map<TableBucket, ReadyWriteBatch> recordsByBucket) {
List<ReadyWriteBatch> writeBatches) {
Map<TableBucket, ReadyWriteBatch> recordsByBucket = new HashMap<>();
writeBatches.forEach(batch -> recordsByBucket.put(batch.tableBucket(), batch));
Comment thread
platinumhamburg marked this conversation as resolved.
long startTime = System.currentTimeMillis();
gateway.putKv(request)
.whenComplete(
(putKvResponse, e) -> {
writerMetricGroup.setSendLatencyInMs(
System.currentTimeMillis() - startTime);
if (e != null) {
handleWriteRequestException(e, recordsByBucket);
handleWriteRequestException(e, writeBatches);
} else {
handlePutKvResponse(putKvResponse, tableId, recordsByBucket);
}
Expand Down Expand Up @@ -530,14 +530,13 @@ private void handlePutKvResponse(
metadataUpdater.invalidPhysicalTableBucketMeta(invalidMetadataTablesSet);
}

private void handleWriteRequestException(
Throwable t, Map<TableBucket, ReadyWriteBatch> recordsByBucket) {
private void handleWriteRequestException(Throwable t, List<ReadyWriteBatch> writeBatches) {
ApiError error = ApiError.fromThrowable(t);

// if batch failed because of retrievable exception, we need to retry send all those
// batches.
Set<PhysicalTablePath> invalidMetadataTablesSet = new HashSet<>();
for (ReadyWriteBatch batch : recordsByBucket.values()) {
for (ReadyWriteBatch batch : writeBatches) {
Set<PhysicalTablePath> invalidMetadataTables = handleWriteBatchException(batch, error);
invalidMetadataTablesSet.addAll(invalidMetadataTables);
}
Expand Down
190 changes: 182 additions & 8 deletions fluss-client/src/test/java/org/apache/fluss/client/write/SenderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -559,7 +564,7 @@ void testCorrectHandlingOfOutOfOrderResponses() throws Exception {

sender1.runOnce(); // receive response 1.
Deque<WriteBatch> 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);
Expand Down Expand Up @@ -629,7 +634,7 @@ void testCorrectHandlingOfOutOfOrderResponsesWhenSecondSucceeds() throws Excepti
assertThat(future2.get()).isNull();
assertThat(future1.isDone()).isFalse();
Deque<WriteBatch> 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));
Expand Down Expand Up @@ -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<Exception> firstFuture = new CompletableFuture<>();
CompletableFuture<Exception> 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);
Expand Down Expand Up @@ -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<Exception> firstFuture = new CompletableFuture<>();
CompletableFuture<Exception> 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<Exception> future1 = new CompletableFuture<>();
Expand Down Expand Up @@ -998,6 +1116,14 @@ private TestingMetadataUpdater initializeMetadataUpdater() {
return new TestingMetadataUpdater(tableInfos);
}

private void resetTableInfosWith(TableInfo tableInfo) {
Map<TablePath, TableInfo> 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);
Expand All @@ -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(),
Expand All @@ -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,
Expand Down Expand Up @@ -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<Long> pendingWriteRequestTableIds(TableBucket referenceBucket) {
List<Long> 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);
}
Comment on lines +1226 to +1232

private void finishIdempotentProduceLogRequest(
int batchSequence, TableBucket tb, int index, ProduceLogResponse response) {
TestTabletServerGateway gateway =
Expand Down
Loading
Loading