This is an automated email from the ASF dual-hosted git repository.
swuferhong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new f01a4ac2e [client] Abort write batch if table id changes. (#2747)
f01a4ac2e is described below
commit f01a4ac2e84976d261a6b95f280a3e31c92fb7a9
Author: Hongshun Wang <[email protected]>
AuthorDate: Tue Jun 2 09:29:44 2026 +0800
[client] Abort write batch if table id changes. (#2747)
* [client] Abort write batch if table id changes
* modified based on advice.
---
.../client/write/AbstractRowLogWriteBatch.java | 3 +-
.../fluss/client/write/ArrowLogWriteBatch.java | 3 +-
.../fluss/client/write/CompactedLogWriteBatch.java | 2 +
.../fluss/client/write/IndexedLogWriteBatch.java | 2 +
.../apache/fluss/client/write/KvWriteBatch.java | 3 +-
.../fluss/client/write/RecordAccumulator.java | 147 ++++++++++++++-------
.../java/org/apache/fluss/client/write/Sender.java | 2 +-
.../org/apache/fluss/client/write/WriteBatch.java | 9 +-
.../apache/fluss/client/write/WriterClient.java | 2 +-
.../client/metadata/TestingMetadataUpdater.java | 5 +
.../client/utils/ClientRpcMessageUtilsTest.java | 1 +
.../fluss/client/write/ArrowLogWriteBatchTest.java | 3 +
.../client/write/CompactedLogWriteBatchTest.java | 19 ++-
.../client/write/IndexedLogWriteBatchTest.java | 19 ++-
.../fluss/client/write/KvWriteBatchTest.java | 2 +
.../org/apache/fluss/client/write/SenderTest.java | 46 ++++++-
.../fluss/exception/TableNotExistException.java | 4 +
17 files changed, 197 insertions(+), 75 deletions(-)
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/write/AbstractRowLogWriteBatch.java
b/fluss-client/src/main/java/org/apache/fluss/client/write/AbstractRowLogWriteBatch.java
index 3c61cce82..fcb71764a 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/write/AbstractRowLogWriteBatch.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/write/AbstractRowLogWriteBatch.java
@@ -44,13 +44,14 @@ abstract class AbstractRowLogWriteBatch<R> extends
WriteBatch {
private final String buildErrorMessage;
protected AbstractRowLogWriteBatch(
+ long tableId,
int bucketId,
PhysicalTablePath physicalTablePath,
long createdMs,
AbstractPagedOutputView outputView,
MemoryLogRecordsRowBuilder<R> recordsBuilder,
String buildErrorMessage) {
- super(bucketId, physicalTablePath, createdMs);
+ super(tableId, bucketId, physicalTablePath, createdMs);
this.outputView = outputView;
this.recordsBuilder = recordsBuilder;
this.buildErrorMessage = buildErrorMessage;
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/write/ArrowLogWriteBatch.java
b/fluss-client/src/main/java/org/apache/fluss/client/write/ArrowLogWriteBatch.java
index bee407cf0..a800666bc 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/write/ArrowLogWriteBatch.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/write/ArrowLogWriteBatch.java
@@ -52,6 +52,7 @@ public class ArrowLogWriteBatch extends WriteBatch {
private final AbstractPagedOutputView outputView;
public ArrowLogWriteBatch(
+ long tableId,
int bucketId,
PhysicalTablePath physicalTablePath,
int schemaId,
@@ -59,7 +60,7 @@ public class ArrowLogWriteBatch extends WriteBatch {
AbstractPagedOutputView outputView,
long createdMs,
@Nullable LogRecordBatchStatisticsCollector statisticsCollector) {
- super(bucketId, physicalTablePath, createdMs);
+ super(tableId, bucketId, physicalTablePath, createdMs);
this.outputView = outputView;
this.recordsBuilder =
MemoryLogRecordsArrowBuilder.builder(
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/write/CompactedLogWriteBatch.java
b/fluss-client/src/main/java/org/apache/fluss/client/write/CompactedLogWriteBatch.java
index 19366a4dc..9b7cee9a7 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/write/CompactedLogWriteBatch.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/write/CompactedLogWriteBatch.java
@@ -40,6 +40,7 @@ import static
org.apache.fluss.utils.Preconditions.checkArgument;
public final class CompactedLogWriteBatch extends
AbstractRowLogWriteBatch<CompactedRow> {
public CompactedLogWriteBatch(
+ long tableId,
int bucketId,
PhysicalTablePath physicalTablePath,
int schemaId,
@@ -47,6 +48,7 @@ public final class CompactedLogWriteBatch extends
AbstractRowLogWriteBatch<Compa
AbstractPagedOutputView outputView,
long createdMs) {
super(
+ tableId,
bucketId,
physicalTablePath,
createdMs,
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/write/IndexedLogWriteBatch.java
b/fluss-client/src/main/java/org/apache/fluss/client/write/IndexedLogWriteBatch.java
index c70dd83a8..dc408a384 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/write/IndexedLogWriteBatch.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/write/IndexedLogWriteBatch.java
@@ -39,6 +39,7 @@ import static
org.apache.fluss.utils.Preconditions.checkArgument;
public final class IndexedLogWriteBatch extends
AbstractRowLogWriteBatch<IndexedRow> {
public IndexedLogWriteBatch(
+ long tableId,
int bucketId,
PhysicalTablePath physicalTablePath,
int schemaId,
@@ -46,6 +47,7 @@ public final class IndexedLogWriteBatch extends
AbstractRowLogWriteBatch<Indexed
AbstractPagedOutputView outputView,
long createdMs) {
super(
+ tableId,
bucketId,
physicalTablePath,
createdMs,
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/write/KvWriteBatch.java
b/fluss-client/src/main/java/org/apache/fluss/client/write/KvWriteBatch.java
index a72d36717..814df03f7 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/write/KvWriteBatch.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/write/KvWriteBatch.java
@@ -55,6 +55,7 @@ public class KvWriteBatch extends WriteBatch {
private final MergeMode mergeMode;
public KvWriteBatch(
+ long tableId,
int bucketId,
PhysicalTablePath physicalTablePath,
int schemaId,
@@ -64,7 +65,7 @@ public class KvWriteBatch extends WriteBatch {
@Nullable int[] targetColumns,
MergeMode mergeMode,
long createdMs) {
- super(bucketId, physicalTablePath, createdMs);
+ super(tableId, bucketId, physicalTablePath, createdMs);
this.outputView = outputView;
this.recordsBuilder =
KvRecordBatchBuilder.builder(schemaId, writeLimit, outputView,
kvFormat);
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/write/RecordAccumulator.java
b/fluss-client/src/main/java/org/apache/fluss/client/write/RecordAccumulator.java
index 80f9b75d1..7a0e0f5dd 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/write/RecordAccumulator.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/write/RecordAccumulator.java
@@ -25,6 +25,7 @@ import org.apache.fluss.cluster.Cluster;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.exception.FlussRuntimeException;
+import org.apache.fluss.exception.TableNotExistException;
import org.apache.fluss.memory.LazyMemorySegmentPool;
import org.apache.fluss.memory.MemorySegment;
import org.apache.fluss.memory.PreAllocatedPagedOutputView;
@@ -316,16 +317,20 @@ public final class RecordAccumulator {
}
/** Abort all incomplete batches (whether they have been sent or not). */
- public void abortBatches(final Exception reason) {
+ public void abortAllBatches(final Exception reason) {
for (WriteBatch batch : incomplete.copyAll()) {
- Deque<WriteBatch> dq = getDeque(batch.physicalTablePath(),
batch.bucketId());
- synchronized (dq) {
- batch.abortRecordAppends();
- dq.remove(batch);
- }
- batch.abort(reason);
- deallocate(batch);
+ abortBatch(reason, batch);
+ }
+ }
+
+ private void abortBatch(final Exception reason, WriteBatch batch) {
+ Deque<WriteBatch> dq = getDeque(batch.physicalTablePath(),
batch.bucketId());
+ synchronized (dq) {
+ batch.abortRecordAppends();
+ dq.remove(batch);
}
+ batch.abort(reason);
+ deallocate(batch);
}
/** Get the deque for the given table-bucket, creating it if necessary. */
@@ -620,6 +625,7 @@ public final class RecordAccumulator {
case COMPACTED_KV:
case INDEXED_KV:
return new KvWriteBatch(
+ tableInfo.getTableId(),
bucketId,
physicalTablePath,
tableInfo.getSchemaId(),
@@ -645,6 +651,7 @@ public final class RecordAccumulator {
tableInfo.getRowType(),
tableInfo.getStatsIndexMapping());
}
return new ArrowLogWriteBatch(
+ tableInfo.getTableId(),
bucketId,
physicalTablePath,
tableInfo.getSchemaId(),
@@ -655,6 +662,7 @@ public final class RecordAccumulator {
case COMPACTED_LOG:
return new CompactedLogWriteBatch(
+ tableInfo.getTableId(),
bucketId,
physicalTablePath,
schemaId,
@@ -664,6 +672,7 @@ public final class RecordAccumulator {
case INDEXED_LOG:
return new IndexedLogWriteBatch(
+ tableInfo.getTableId(),
bucketId,
physicalTablePath,
tableInfo.getSchemaId(),
@@ -723,57 +732,103 @@ public final class RecordAccumulator {
}
final WriteBatch batch;
+ List<WriteBatch> staleBatches = null;
+ long oldStaleTableId = -1L;
synchronized (deque) {
WriteBatch first = deque.peekFirst();
if (first == null) {
continue;
}
- // TODO retry back off check.
-
- if (size + first.estimatedSizeInBytes() > maxSize &&
!ready.isEmpty()) {
- // there is a rare case that a single batch size is larger
than the request size
- // due to compression; in this case we will still
eventually send this batch in
- // a single request.
- break;
+ if (tableBucket.getTableId() != first.tableId()) {
+ // Table has been dropped and re-created with a new table
id. Drain ALL
+ // consecutive head batches that belong to the old table
instance in one
+ // pass under the lock, then abort them outside the lock
with a single
+ // aggregated WARN line.
+ oldStaleTableId = first.tableId();
+ staleBatches = new ArrayList<>();
+ while (first != null
+ && first.tableId() == oldStaleTableId
+ && first.tableId() != tableBucket.getTableId()) {
+ staleBatches.add(deque.pollFirst());
+ first = deque.peekFirst();
+ }
+ batch = null;
} else {
- if (shouldStopDrainBatchesForBucket(first, tableBucket)) {
+ // TODO retry back off check.
+
+ if (size + first.estimatedSizeInBytes() > maxSize &&
!ready.isEmpty()) {
+ // there is a rare case that a single batch size is
larger than the
+ // request size due to compression; in this case we
will still
+ // eventually send this batch in a single request.
+ break;
+ } else if (shouldStopDrainBatchesForBucket(first,
tableBucket)) {
// Buckets are independent — skip this one, keep
draining others.
continue;
}
+
+ batch = deque.pollFirst();
+ long writerId =
+ idempotenceManager.idempotenceEnabled()
+ ? idempotenceManager.writerId()
+ : NO_WRITER_ID;
+ if (writerId != NO_WRITER_ID && !batch.hasBatchSequence())
{
+ // If writer id of the bucket do not match the latest
one of writer,
+ // we update it and reset the batch sequence. This
should be only done when
+ // all
+ // its in-flight batches have completed. This is
guarantee in
+ // `shouldStopDrainBatchesForBucket`.
+ idempotenceManager.maybeUpdateWriterId(tableBucket);
+
+ // If the batch already has an assigned batch
sequence, then we should not
+ // change writer id and batch sequence, since this may
introduce
+ // duplicates. In particular, the previous attempt may
actually have been
+ // accepted, and if we change writer id and sequence
here, this attempt
+ // will also be accepted, causing a duplicate.
+ //
+ // Additionally, we update the next batch sequence
bound for the table
+ // bucket,
+ // and also have the writerStateManager track the
batch to ensure
+ // that sequence ordering is maintained even if we
receive out of order
+ // responses.
+ batch.setWriterState(
+ writerId,
idempotenceManager.nextSequence(tableBucket));
+ idempotenceManager.incrementBatchSequence(tableBucket);
+ LOG.debug(
+ "Assigner writerId {} to batch with batch
sequence {} being sent to table bucket {}",
+ writerId,
+ batch.batchSequence(),
+ tableBucket);
+ idempotenceManager.addInFlightBatch(batch,
tableBucket);
+ }
}
+ }
- batch = deque.pollFirst();
- long writerId =
- idempotenceManager.idempotenceEnabled()
- ? idempotenceManager.writerId()
- : NO_WRITER_ID;
- if (writerId != NO_WRITER_ID && !batch.hasBatchSequence()) {
- // If writer id of the bucket do not match the latest one
of writer,
- // we update it and reset the batch sequence. This should
be only done when all
- // its in-flight batches have completed. This is guarantee
in
- // `shouldStopDrainBatchesForBucket`.
- idempotenceManager.maybeUpdateWriterId(tableBucket);
-
- // If the batch already has an assigned batch sequence,
then we should not
- // change writer id and batch sequence, since this may
introduce
- // duplicates. In particular, the previous attempt may
actually have been
- // accepted, and if we change writer id and sequence here,
this attempt
- // will also be accepted, causing a duplicate.
- //
- // Additionally, we update the next batch sequence bound
for the table bucket,
- // and also have the writerStateManager track the batch to
ensure
- // that sequence ordering is maintained even if we receive
out of order
- // responses.
- batch.setWriterState(writerId,
idempotenceManager.nextSequence(tableBucket));
- idempotenceManager.incrementBatchSequence(tableBucket);
- LOG.debug(
- "Assigner writerId {} to batch with batch sequence
{} being sent to table bucket {}",
- writerId,
- batch.batchSequence(),
- tableBucket);
- idempotenceManager.addInFlightBatch(batch, tableBucket);
+ // Abort stale batches *outside* the deque lock so user callbacks
(alien methods)
+ // and memory deallocation never run while holding it. Aggregate
to a single WARN.
+ if (staleBatches != null) {
+ LOG.warn(
+ "Table {} has been dropped and re-created with a new
table ID. "
+ + "Old ID: {}, New ID: {}. Aborting {} pending
batches for the old table instance.",
+ physicalTablePath,
+ oldStaleTableId,
+ tableBucket.getTableId(),
+ staleBatches.size());
+ TableNotExistException reason =
+ new TableNotExistException(
+ String.format(
+ "Table '%s' has been dropped and
re-created with a new table ID (old: %d, new: %d). "
+ + "Further writes to the old
table instance cannot proceed. "
+ + "Please recreate the writer
with the new table metadata.",
+ physicalTablePath,
+ oldStaleTableId,
+ tableBucket.getTableId()),
+ null,
+ true);
+ for (WriteBatch staleBatch : staleBatches) {
+ abortBatch(reason, staleBatch);
}
+ continue;
}
// the rest of the work by processing outside the lock close() is
particularly expensive
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 7b456d4bd..a8e62cd37 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
@@ -299,7 +299,7 @@ public class Sender implements Runnable {
private void maybeAbortBatches(Exception exception) {
if (accumulator.hasIncomplete()) {
LOG.error("Aborting write batches due to fatal error", exception);
- accumulator.abortBatches(exception);
+ accumulator.abortAllBatches(exception);
}
}
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/write/WriteBatch.java
b/fluss-client/src/main/java/org/apache/fluss/client/write/WriteBatch.java
index cb83943cf..9702bb543 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/write/WriteBatch.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/write/WriteBatch.java
@@ -46,6 +46,7 @@ public abstract class WriteBatch {
private final long createdMs;
private final PhysicalTablePath physicalTablePath;
private final RequestFuture requestFuture;
+ private final long tableId;
private final int bucketId;
protected final List<WriteCallback> callbacks = new ArrayList<>();
@@ -55,9 +56,11 @@ public abstract class WriteBatch {
protected int recordCount;
private long drainedMs;
- public WriteBatch(int bucketId, PhysicalTablePath physicalTablePath, long
createdMs) {
+ public WriteBatch(
+ long tableId, int bucketId, PhysicalTablePath physicalTablePath,
long createdMs) {
this.physicalTablePath = physicalTablePath;
this.createdMs = createdMs;
+ this.tableId = tableId;
this.bucketId = bucketId;
this.requestFuture = new RequestFuture();
this.recordCount = 0;
@@ -157,6 +160,10 @@ public abstract class WriteBatch {
return bucketId;
}
+ public long tableId() {
+ return tableId;
+ }
+
public PhysicalTablePath physicalTablePath() {
return physicalTablePath;
}
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/write/WriterClient.java
b/fluss-client/src/main/java/org/apache/fluss/client/write/WriterClient.java
index 351c7c065..b4c96b0ac 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/write/WriterClient.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/write/WriterClient.java
@@ -227,7 +227,7 @@ public class WriterClient {
private void maybeAbortBatches(Throwable t) {
if (accumulator.hasIncomplete()) {
LOG.error("Aborting all pending write batches due to fatal error",
t);
- accumulator.abortBatches(toException(t));
+ accumulator.abortAllBatches(toException(t));
}
}
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/metadata/TestingMetadataUpdater.java
b/fluss-client/src/test/java/org/apache/fluss/client/metadata/TestingMetadataUpdater.java
index 6c67d297e..303e45318 100644
---
a/fluss-client/src/test/java/org/apache/fluss/client/metadata/TestingMetadataUpdater.java
+++
b/fluss-client/src/test/java/org/apache/fluss/client/metadata/TestingMetadataUpdater.java
@@ -87,6 +87,11 @@ public class TestingMetadataUpdater extends MetadataUpdater {
}
}
+ public void updateTableInfos(Map<TablePath, TableInfo> tableInfos) {
+ initializeCluster(
+ cluster.getCoordinatorServer(),
cluster.getAliveTabletServerList(), tableInfos);
+ }
+
/**
* Create a builder for constructing TestingMetadataUpdater with custom
gateways.
*
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/utils/ClientRpcMessageUtilsTest.java
b/fluss-client/src/test/java/org/apache/fluss/client/utils/ClientRpcMessageUtilsTest.java
index 5b1ca58c7..3ed17da7d 100644
---
a/fluss-client/src/test/java/org/apache/fluss/client/utils/ClientRpcMessageUtilsTest.java
+++
b/fluss-client/src/test/java/org/apache/fluss/client/utils/ClientRpcMessageUtilsTest.java
@@ -131,6 +131,7 @@ class ClientRpcMessageUtilsTest {
PreAllocatedPagedOutputView outputView =
new
PreAllocatedPagedOutputView(Collections.singletonList(segment));
return new KvWriteBatch(
+ DATA1_TABLE_ID_PK,
bucketId,
PhysicalTablePath.of(DATA1_TABLE_PATH_PK),
DATA1_TABLE_INFO_PK.getSchemaId(),
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/write/ArrowLogWriteBatchTest.java
b/fluss-client/src/test/java/org/apache/fluss/client/write/ArrowLogWriteBatchTest.java
index 07c4aa338..f87f0c01a 100644
---
a/fluss-client/src/test/java/org/apache/fluss/client/write/ArrowLogWriteBatchTest.java
+++
b/fluss-client/src/test/java/org/apache/fluss/client/write/ArrowLogWriteBatchTest.java
@@ -126,6 +126,7 @@ public class ArrowLogWriteBatchTest {
TableBucket tb = new TableBucket(DATA1_TABLE_ID, bucketId);
ArrowLogWriteBatch arrowLogWriteBatch =
new ArrowLogWriteBatch(
+ tb.getTableId(),
tb.getBucket(),
DATA1_PHYSICAL_TABLE_PATH,
DATA1_TABLE_INFO.getSchemaId(),
@@ -206,6 +207,7 @@ public class ArrowLogWriteBatchTest {
ArrowLogWriteBatch arrowLogWriteBatch =
new ArrowLogWriteBatch(
+ tb.getTableId(),
tb.getBucket(),
DATA1_PHYSICAL_TABLE_PATH,
DATA1_TABLE_INFO.getSchemaId(),
@@ -302,6 +304,7 @@ public class ArrowLogWriteBatchTest {
private ArrowLogWriteBatch createArrowLogWriteBatch(TableBucket tb, int
maxSizeInBytes) {
return new ArrowLogWriteBatch(
+ tb.getTableId(),
tb.getBucket(),
DATA1_PHYSICAL_TABLE_PATH,
DATA1_TABLE_INFO.getSchemaId(),
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/write/CompactedLogWriteBatchTest.java
b/fluss-client/src/test/java/org/apache/fluss/client/write/CompactedLogWriteBatchTest.java
index 4f8b37919..140058eca 100644
---
a/fluss-client/src/test/java/org/apache/fluss/client/write/CompactedLogWriteBatchTest.java
+++
b/fluss-client/src/test/java/org/apache/fluss/client/write/CompactedLogWriteBatchTest.java
@@ -67,7 +67,6 @@ public class CompactedLogWriteBatchTest {
CompactedLogWriteBatch logProducerBatch =
createLogWriteBatch(
new TableBucket(DATA1_TABLE_ID, bucketId),
- 0L,
writeLimit,
MemorySegment.allocateHeapMemory(writeLimit));
@@ -89,7 +88,7 @@ public class CompactedLogWriteBatchTest {
void testToBytes() throws Exception {
int bucketId = 0;
CompactedLogWriteBatch logProducerBatch =
- createLogWriteBatch(new TableBucket(DATA1_TABLE_ID, bucketId),
0L);
+ createLogWriteBatch(new TableBucket(DATA1_TABLE_ID, bucketId));
boolean appendResult = logProducerBatch.tryAppend(createWriteRecord(),
newWriteCallback());
assertThat(appendResult).isTrue();
@@ -105,7 +104,7 @@ public class CompactedLogWriteBatchTest {
void testCompleteTwice() throws Exception {
int bucketId = 0;
CompactedLogWriteBatch logWriteBatch =
- createLogWriteBatch(new TableBucket(DATA1_TABLE_ID, bucketId),
0L);
+ createLogWriteBatch(new TableBucket(DATA1_TABLE_ID, bucketId));
boolean appendResult = logWriteBatch.tryAppend(createWriteRecord(),
newWriteCallback());
assertThat(appendResult).isTrue();
@@ -120,7 +119,7 @@ public class CompactedLogWriteBatchTest {
void testFailedTwice() throws Exception {
int bucketId = 0;
CompactedLogWriteBatch logWriteBatch =
- createLogWriteBatch(new TableBucket(DATA1_TABLE_ID, bucketId),
0L);
+ createLogWriteBatch(new TableBucket(DATA1_TABLE_ID, bucketId));
boolean appendResult = logWriteBatch.tryAppend(createWriteRecord(),
newWriteCallback());
assertThat(appendResult).isTrue();
@@ -135,7 +134,7 @@ public class CompactedLogWriteBatchTest {
void testClose() throws Exception {
int bucketId = 0;
CompactedLogWriteBatch logProducerBatch =
- createLogWriteBatch(new TableBucket(DATA1_TABLE_ID, bucketId),
0L);
+ createLogWriteBatch(new TableBucket(DATA1_TABLE_ID, bucketId));
boolean appendResult = logProducerBatch.tryAppend(createWriteRecord(),
newWriteCallback());
assertThat(appendResult).isTrue();
@@ -153,7 +152,6 @@ public class CompactedLogWriteBatchTest {
CompactedLogWriteBatch logProducerBatch =
createLogWriteBatch(
new TableBucket(DATA1_TABLE_ID, bucketId),
- 0L,
writeLimit,
MemorySegment.allocateHeapMemory(writeLimit));
@@ -202,15 +200,14 @@ public class CompactedLogWriteBatchTest {
DATA1_TABLE_INFO, DATA1_PHYSICAL_TABLE_PATH, row, null);
}
- private CompactedLogWriteBatch createLogWriteBatch(TableBucket tb, long
baseLogOffset)
- throws Exception {
- return createLogWriteBatch(
- tb, baseLogOffset, Integer.MAX_VALUE,
MemorySegment.allocateHeapMemory(1000));
+ private CompactedLogWriteBatch createLogWriteBatch(TableBucket tb) throws
Exception {
+ return createLogWriteBatch(tb, Integer.MAX_VALUE,
MemorySegment.allocateHeapMemory(1000));
}
private CompactedLogWriteBatch createLogWriteBatch(
- TableBucket tb, long baseLogOffset, int writeLimit, MemorySegment
memorySegment) {
+ TableBucket tb, int writeLimit, MemorySegment memorySegment) {
return new CompactedLogWriteBatch(
+ tb.getTableId(),
tb.getBucket(),
DATA1_PHYSICAL_TABLE_PATH,
DATA1_TABLE_INFO.getSchemaId(),
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/write/IndexedLogWriteBatchTest.java
b/fluss-client/src/test/java/org/apache/fluss/client/write/IndexedLogWriteBatchTest.java
index 683013b89..331be4209 100644
---
a/fluss-client/src/test/java/org/apache/fluss/client/write/IndexedLogWriteBatchTest.java
+++
b/fluss-client/src/test/java/org/apache/fluss/client/write/IndexedLogWriteBatchTest.java
@@ -70,7 +70,6 @@ public class IndexedLogWriteBatchTest {
IndexedLogWriteBatch logProducerBatch =
createLogWriteBatch(
new TableBucket(DATA1_TABLE_ID, bucketId),
- 0L,
writeLimit,
MemorySegment.allocateHeapMemory(writeLimit));
@@ -92,7 +91,7 @@ public class IndexedLogWriteBatchTest {
void testToBytes() throws Exception {
int bucketId = 0;
IndexedLogWriteBatch logProducerBatch =
- createLogWriteBatch(new TableBucket(DATA1_TABLE_ID, bucketId),
0L);
+ createLogWriteBatch(new TableBucket(DATA1_TABLE_ID, bucketId));
boolean appendResult = logProducerBatch.tryAppend(createWriteRecord(),
newWriteCallback());
assertThat(appendResult).isTrue();
@@ -109,7 +108,7 @@ public class IndexedLogWriteBatchTest {
void testCompleteTwice() throws Exception {
int bucketId = 0;
IndexedLogWriteBatch logWriteBatch =
- createLogWriteBatch(new TableBucket(DATA1_TABLE_ID, bucketId),
0L);
+ createLogWriteBatch(new TableBucket(DATA1_TABLE_ID, bucketId));
boolean appendResult = logWriteBatch.tryAppend(createWriteRecord(),
newWriteCallback());
assertThat(appendResult).isTrue();
@@ -124,7 +123,7 @@ public class IndexedLogWriteBatchTest {
void testFailedTwice() throws Exception {
int bucketId = 0;
IndexedLogWriteBatch logWriteBatch =
- createLogWriteBatch(new TableBucket(DATA1_TABLE_ID, bucketId),
0L);
+ createLogWriteBatch(new TableBucket(DATA1_TABLE_ID, bucketId));
boolean appendResult = logWriteBatch.tryAppend(createWriteRecord(),
newWriteCallback());
assertThat(appendResult).isTrue();
@@ -139,7 +138,7 @@ public class IndexedLogWriteBatchTest {
void testClose() throws Exception {
int bucketId = 0;
IndexedLogWriteBatch logProducerBatch =
- createLogWriteBatch(new TableBucket(DATA1_TABLE_ID, bucketId),
0L);
+ createLogWriteBatch(new TableBucket(DATA1_TABLE_ID, bucketId));
boolean appendResult = logProducerBatch.tryAppend(createWriteRecord(),
newWriteCallback());
assertThat(appendResult).isTrue();
@@ -157,7 +156,6 @@ public class IndexedLogWriteBatchTest {
IndexedLogWriteBatch logProducerBatch =
createLogWriteBatch(
new TableBucket(DATA1_TABLE_ID, bucketId),
- 0L,
writeLimit,
MemorySegment.allocateHeapMemory(writeLimit));
@@ -205,15 +203,14 @@ public class IndexedLogWriteBatchTest {
return WriteRecord.forIndexedAppend(DATA1_TABLE_INFO,
DATA1_PHYSICAL_TABLE_PATH, row, null);
}
- private IndexedLogWriteBatch createLogWriteBatch(TableBucket tb, long
baseLogOffset)
- throws Exception {
- return createLogWriteBatch(
- tb, baseLogOffset, Integer.MAX_VALUE,
MemorySegment.allocateHeapMemory(1000));
+ private IndexedLogWriteBatch createLogWriteBatch(TableBucket tb) throws
Exception {
+ return createLogWriteBatch(tb, Integer.MAX_VALUE,
MemorySegment.allocateHeapMemory(1000));
}
private IndexedLogWriteBatch createLogWriteBatch(
- TableBucket tb, long baseLogOffset, int writeLimit, MemorySegment
memorySegment) {
+ TableBucket tb, int writeLimit, MemorySegment memorySegment) {
return new IndexedLogWriteBatch(
+ tb.getTableId(),
tb.getBucket(),
DATA1_PHYSICAL_TABLE_PATH,
DATA1_TABLE_INFO.getSchemaId(),
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/write/KvWriteBatchTest.java
b/fluss-client/src/test/java/org/apache/fluss/client/write/KvWriteBatchTest.java
index e2340d170..7b61976a0 100644
---
a/fluss-client/src/test/java/org/apache/fluss/client/write/KvWriteBatchTest.java
+++
b/fluss-client/src/test/java/org/apache/fluss/client/write/KvWriteBatchTest.java
@@ -221,6 +221,7 @@ class KvWriteBatchTest {
PreAllocatedPagedOutputView outputView =
new
PreAllocatedPagedOutputView(Collections.singletonList(memorySegment));
return new KvWriteBatch(
+ tb.getTableId(),
tb.getBucket(),
PhysicalTablePath.of(DATA1_TABLE_PATH_PK),
DATA1_TABLE_INFO_PK.getSchemaId(),
@@ -316,6 +317,7 @@ class KvWriteBatchTest {
new PreAllocatedPagedOutputView(
Collections.singletonList(memoryPool.nextSegment()));
return new KvWriteBatch(
+ tb.getTableId(),
tb.getBucket(),
PhysicalTablePath.of(DATA1_TABLE_PATH_PK),
DATA1_TABLE_INFO_PK.getSchemaId(),
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 2c4d30e96..be12bb36e 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,7 @@ import org.apache.fluss.cluster.ServerNode;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.config.MemorySize;
+import org.apache.fluss.exception.TableNotExistException;
import org.apache.fluss.exception.TimeoutException;
import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.TableBucket;
@@ -62,12 +63,15 @@ 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_ID;
import static org.apache.fluss.record.TestData.DATA1_TABLE_ID_PK;
import static org.apache.fluss.record.TestData.DATA1_TABLE_INFO;
import static org.apache.fluss.record.TestData.DATA1_TABLE_INFO_PK;
import static org.apache.fluss.record.TestData.DATA1_TABLE_PATH;
import static org.apache.fluss.record.TestData.DATA1_TABLE_PATH_PK;
+import static org.apache.fluss.record.TestData.DATA2_TABLE_ID;
+import static org.apache.fluss.record.TestData.DEFAULT_REMOTE_DATA_DIR;
import static org.apache.fluss.rpc.protocol.Errors.SCHEMA_NOT_EXIST;
import static
org.apache.fluss.server.utils.ServerRpcMessageUtils.getProduceLogData;
import static
org.apache.fluss.server.utils.ServerRpcMessageUtils.makeProduceLogResponse;
@@ -880,6 +884,40 @@ final class SenderTest {
assertThat(future.get()).isNull();
}
+ @Test
+ void testSendWhenTableIdChanges() throws Exception {
+ CompletableFuture<Exception> future1 = new CompletableFuture<>();
+ appendToAccumulator(tb1, row(1, "a"), (tb, leo, e) ->
future1.complete(e));
+ TableInfo newTableInfo =
+ TableInfo.of(
+ DATA1_TABLE_PATH,
+ DATA2_TABLE_ID,
+ 1,
+ DATA1_TABLE_DESCRIPTOR,
+ DEFAULT_REMOTE_DATA_DIR,
+ System.currentTimeMillis(),
+ System.currentTimeMillis());
+ TableBucket newTableBucket = new
TableBucket(newTableInfo.getTableId(), tb1.getBucket());
+
+
metadataUpdater.updateTableInfos(Collections.singletonMap(DATA1_TABLE_PATH,
newTableInfo));
+ sender.runOnce();
+ Exception exception = future1.get();
+ assertThat(exception).isNotNull();
+
assertThat(exception).isExactlyInstanceOf(TableNotExistException.class);
+ assertThat(exception.getMessage())
+ .contains(
+ String.format(
+ "Table '%s' has been dropped and re-created
with a new table ID (old: %s, new: %s)",
+ DATA1_TABLE_PATH, DATA1_TABLE_ID,
newTableInfo.getTableId()));
+
+ CompletableFuture<Exception> future2 = new CompletableFuture<>();
+ appendToAccumulator(
+ newTableInfo, newTableBucket, row(1, "a"), (tb, leo, e) ->
future2.complete(e));
+ sender.runOnce();
+ finishRequest(newTableBucket, 0,
createProduceLogResponse(newTableBucket, 0, 1));
+ assertThat(future2.get()).isNull();
+ }
+
private TestingMetadataUpdater initializeMetadataUpdater() {
Map<TablePath, TableInfo> tableInfos = new HashMap<>();
tableInfos.put(DATA1_TABLE_PATH, DATA1_TABLE_INFO);
@@ -889,8 +927,14 @@ final class SenderTest {
private void appendToAccumulator(TableBucket tb, GenericRow row,
WriteCallback writeCallback)
throws Exception {
+ appendToAccumulator(DATA1_TABLE_INFO, tb, row, writeCallback);
+ }
+
+ private void appendToAccumulator(
+ TableInfo tableInfo, TableBucket tb, GenericRow row, WriteCallback
writeCallback)
+ throws Exception {
accumulator.append(
- WriteRecord.forArrowAppend(DATA1_TABLE_INFO,
DATA1_PHYSICAL_TABLE_PATH, row, null),
+ WriteRecord.forArrowAppend(tableInfo,
DATA1_PHYSICAL_TABLE_PATH, row, null),
writeCallback,
metadataUpdater.getCluster(),
tb.getBucket(),
diff --git
a/fluss-common/src/main/java/org/apache/fluss/exception/TableNotExistException.java
b/fluss-common/src/main/java/org/apache/fluss/exception/TableNotExistException.java
index b101a71bd..7c1110690 100644
---
a/fluss-common/src/main/java/org/apache/fluss/exception/TableNotExistException.java
+++
b/fluss-common/src/main/java/org/apache/fluss/exception/TableNotExistException.java
@@ -33,4 +33,8 @@ public class TableNotExistException extends ApiException {
public TableNotExistException(String message, Throwable cause) {
super(message, cause);
}
+
+ public TableNotExistException(String message, Throwable cause, boolean
stackTraceEnabled) {
+ super(message, cause, stackTraceEnabled);
+ }
}
