You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by yu...@apache.org on 2026/09/22 12:29:54 UTC
(fluss) branch main updated: [tiering] Optimize tiering split queue and commit collection (#4457)
This is an automated email from the ASF dual-hosted git repository.
luoyuxia 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 55dc68528 [tiering] Optimize tiering split queue and commit collection (#4457)
55dc68528 is described below
commit 55dc685281dd14c46fdac63dd0140d2d216a3008
Author: Junbo Wang <be...@gmail.com>
AuthorDate: Tue Sep 22 20:29:48 2026 +0800
[tiering] Optimize tiering split queue and commit collection (#4457)
---
.../tiering/committer/TieringCommitOperator.java | 103 +++++++++------------
.../source/enumerator/TieringSourceEnumerator.java | 11 ++-
.../committer/TieringCommitOperatorTest.java | 72 +++++++++++++-
.../TieringSourceEnumeratorSchedulingTest.java | 36 ++++++-
.../enumerator/TieringSourceEnumeratorTest.java | 26 ++++--
5 files changed, 172 insertions(+), 76 deletions(-)
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java
index 477beea11..218af6888 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java
@@ -91,9 +91,8 @@ public class TieringCommitOperator<WriteResult, Committable>
// gateway to send event to flink source coordinator
private final OperatorEventGateway operatorEventGateway;
- // tableid -> write results
- private final Map<Long, List<TableBucketWriteResult<WriteResult>>>
- collectedTableBucketWriteResults;
+ // Table ID -> accumulator for the current tiering round.
+ private final Map<Long, TableWriteResultAccumulator<WriteResult>> tableWriteResultAccumulators;
/**
* The result of one table's commit round, holding the lake committable (nullable for empty
@@ -118,7 +117,7 @@ public class TieringCommitOperator<WriteResult, Committable>
LakeTieringFactory<WriteResult, Committable> lakeTieringFactory) {
this.lakeTieringFactory = lakeTieringFactory;
this.flussTableLakeSnapshotCommitter = new FlussTableLakeSnapshotCommitter(flussConf);
- this.collectedTableBucketWriteResults = new HashMap<>();
+ this.tableWriteResultAccumulators = new HashMap<>();
this.flussConfig = flussConf;
this.lakeTieringConfig = lakeTieringConfig;
this.setup(
@@ -144,19 +143,21 @@ public class TieringCommitOperator<WriteResult, Committable>
TableBucketWriteResult<WriteResult> tableBucketWriteResult = streamRecord.getValue();
TableBucket tableBucket = tableBucketWriteResult.tableBucket();
long tableId = tableBucket.getTableId();
- registerTableBucketWriteResult(tableId, tableBucketWriteResult);
-
- // may collect all write results for the table
- List<TableBucketWriteResult<WriteResult>> committableWriteResults =
- collectTableAllBucketWriteResult(tableId);
+ TableWriteResultAccumulator<WriteResult> accumulator =
+ tableWriteResultAccumulators.computeIfAbsent(
+ tableId,
+ ignored ->
+ new TableWriteResultAccumulator<>(
+ tableBucketWriteResult.numberOfWriteResults()));
+ accumulator.add(tableBucketWriteResult);
- if (committableWriteResults != null) {
+ if (accumulator.isComplete()) {
try {
CommitResult commitResult =
commitWriteResults(
tableId,
tableBucketWriteResult.tablePath(),
- committableWriteResults);
+ accumulator.bucketWriteResults);
// only emit downstream when actual data was written
if (commitResult.committable != null) {
output.collect(
@@ -176,7 +177,7 @@ public class TieringCommitOperator<WriteResult, Committable>
"Fail to commit tiering write result, will try to tier again in next round.",
e);
} finally {
- collectedTableBucketWriteResults.remove(tableId);
+ tableWriteResultAccumulators.remove(tableId);
}
}
}
@@ -380,51 +381,6 @@ public class TieringCommitOperator<WriteResult, Committable>
}
}
- private void registerTableBucketWriteResult(
- long tableId, TableBucketWriteResult<WriteResult> tableBucketWriteResult) {
- collectedTableBucketWriteResults
- .computeIfAbsent(tableId, k -> new ArrayList<>())
- .add(tableBucketWriteResult);
- }
-
- @Nullable
- private List<TableBucketWriteResult<WriteResult>> collectTableAllBucketWriteResult(
- long tableId) {
- Set<TableBucket> collectedBuckets = new HashSet<>();
- Integer numberOfWriteResults = null;
- List<TableBucketWriteResult<WriteResult>> writeResults = new ArrayList<>();
- for (TableBucketWriteResult<WriteResult> tableBucketWriteResult :
- collectedTableBucketWriteResults.get(tableId)) {
- if (!collectedBuckets.add(tableBucketWriteResult.tableBucket())) {
- // it means the write results contain more than two write result
- // for same table, it shouldn't happen, let's throw exception to
- // avoid unexpected behavior
- throw new IllegalStateException(
- String.format(
- "Found duplicate write results for bucket %s of table %s.",
- tableBucketWriteResult.tableBucket(), tableId));
- }
- if (numberOfWriteResults == null) {
- numberOfWriteResults = tableBucketWriteResult.numberOfWriteResults();
- } else {
- // the numberOfWriteResults must be same across tableBucketWriteResults
- checkState(
- numberOfWriteResults == tableBucketWriteResult.numberOfWriteResults(),
- "numberOfWriteResults is not same across TableBucketWriteResults for table %s, got %s and %s.",
- tableId,
- numberOfWriteResults,
- tableBucketWriteResult.numberOfWriteResults());
- }
- writeResults.add(tableBucketWriteResult);
- }
-
- if (numberOfWriteResults != null && writeResults.size() == numberOfWriteResults) {
- return writeResults;
- } else {
- return null;
- }
- }
-
@Override
public void close() throws Exception {
flussTableLakeSnapshotCommitter.close();
@@ -435,4 +391,37 @@ public class TieringCommitOperator<WriteResult, Committable>
connection.close();
}
}
+
+ /** Collects one table's tiering round with constant-time validation of each result. */
+ private static final class TableWriteResultAccumulator<WriteResult> {
+ private final int expectedNumberOfWriteResults;
+ private final Set<TableBucket> collectedBuckets = new HashSet<>();
+ private final List<TableBucketWriteResult<WriteResult>> bucketWriteResults =
+ new ArrayList<>();
+
+ private TableWriteResultAccumulator(int expectedNumberOfWriteResults) {
+ this.expectedNumberOfWriteResults = expectedNumberOfWriteResults;
+ }
+
+ private void add(TableBucketWriteResult<WriteResult> bucketWriteResult) {
+ TableBucket tableBucket = bucketWriteResult.tableBucket();
+ checkState(
+ !collectedBuckets.contains(tableBucket),
+ "Found duplicate write results for bucket %s of table %s.",
+ tableBucket,
+ tableBucket.getTableId());
+ checkState(
+ expectedNumberOfWriteResults == bucketWriteResult.numberOfWriteResults(),
+ "numberOfWriteResults is not same across TableBucketWriteResults for table %s, got %s and %s.",
+ tableBucket.getTableId(),
+ expectedNumberOfWriteResults,
+ bucketWriteResult.numberOfWriteResults());
+ collectedBuckets.add(tableBucket);
+ bucketWriteResults.add(bucketWriteResult);
+ }
+
+ private boolean isComplete() {
+ return bucketWriteResults.size() == expectedNumberOfWriteResults;
+ }
+ }
}
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java
index fe9dcef10..2f30a08ee 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java
@@ -61,6 +61,7 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.Deque;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -69,6 +70,7 @@ import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -108,7 +110,8 @@ public class TieringSourceEnumerator
private final ScheduledExecutorService timerService;
private final SplitEnumeratorMetricGroup enumeratorMetricGroup;
private final long pollTieringTableIntervalMs;
- private final List<TieringSplit> pendingSplits;
+ // Also inspected by asynchronous heartbeat requests.
+ private final Deque<TieringSplit> pendingSplits;
private final Set<Integer> readersAwaitingSplit;
private final Map<Long, Long> tieringTableEpochs;
@@ -158,7 +161,7 @@ public class TieringSourceEnumerator
this.timerService = timerService;
this.enumeratorMetricGroup = context.metricGroup();
this.pollTieringTableIntervalMs = pollTieringTableIntervalMs;
- this.pendingSplits = Collections.synchronizedList(new ArrayList<>());
+ this.pendingSplits = new ConcurrentLinkedDeque<>();
this.readersAwaitingSplit = Collections.synchronizedSet(new TreeSet<>());
this.tieringTableEpochs = new ConcurrentHashMap<>();
this.finishedTables = new ConcurrentHashMap<>();
@@ -410,8 +413,8 @@ public class TieringSourceEnumerator
readersAwaitingSplit.remove(nextAwaitingReader);
continue;
}
- if (!pendingSplits.isEmpty()) {
- TieringSplit tieringSplit = pendingSplits.remove(0);
+ TieringSplit tieringSplit = pendingSplits.pollFirst();
+ if (tieringSplit != null) {
context.assignSplit(tieringSplit, nextAwaitingReader);
LOG.info("Assigning split {} to readers {}", tieringSplit, nextAwaitingReader);
readersAwaitingSplit.remove(nextAwaitingReader);
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperatorTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperatorTest.java
index 2e4d85d18..956fa3cf2 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperatorTest.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperatorTest.java
@@ -69,9 +69,11 @@ class TieringCommitOperatorTest extends FlinkTestBase {
private TieringCommitOperator<TestingWriteResult, TestingCommittable> committerOperator;
private MockOperatorEventGateway mockOperatorEventGateway;
private StreamOperatorParameters<CommittableMessage<TestingCommittable>> parameters;
+ private List<CommittableMessage<TestingCommittable>> output;
@BeforeEach
public void beforeEach() throws Exception {
+ output = new ArrayList<>();
mockOperatorEventGateway = new MockOperatorEventGateway();
MockOperatorEventDispatcher mockOperatorEventDispatcher =
new MockOperatorEventDispatcher(mockOperatorEventGateway);
@@ -79,7 +81,7 @@ class TieringCommitOperatorTest extends FlinkTestBase {
StreamOperatorParametersAdapter.create(
new SourceOperatorStreamTask<String>(new DummyEnvironment()),
new MockStreamConfig(new Configuration(), 1),
- new MockOutput<>(new ArrayList<>()),
+ new MockOutput<>(output),
null,
mockOperatorEventDispatcher,
null);
@@ -98,6 +100,52 @@ class TieringCommitOperatorTest extends FlinkTestBase {
committerOperator.close();
}
+ @Test
+ void testDuplicateBucketWriteResult() throws Exception {
+ TablePath tablePath = TablePath.of("fluss", "duplicate_bucket");
+ long tableId = createTable(tablePath, DEFAULT_PK_TABLE_DESCRIPTOR);
+ TableBucket tableBucket = new TableBucket(tableId, 0);
+ committerOperator.processElement(
+ createTableBucketWriteResultStreamRecord(tablePath, tableBucket, 1, 1, 1, 2));
+
+ // The duplicate would reach the expected count, but must fail instead of committing.
+ assertThatThrownBy(
+ () ->
+ committerOperator.processElement(
+ createTableBucketWriteResultStreamRecord(
+ tablePath, tableBucket, 2, 2, 2, 2)))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage(
+ "Found duplicate write results for bucket %s of table %s.",
+ tableBucket, tableId);
+ verifyNoLakeSnapshot(tablePath);
+ }
+
+ @Test
+ void testInconsistentWriteResultCounts() throws Exception {
+ TablePath tablePath = TablePath.of("fluss", "inconsistent_result_counts");
+ long tableId = createTable(tablePath, DEFAULT_PK_TABLE_DESCRIPTOR);
+ committerOperator.processElement(
+ createTableBucketWriteResultStreamRecord(
+ tablePath, new TableBucket(tableId, 0), 1, 1, 1, 2));
+
+ assertThatThrownBy(
+ () ->
+ committerOperator.processElement(
+ createTableBucketWriteResultStreamRecord(
+ tablePath,
+ new TableBucket(tableId, 1),
+ 2,
+ 2,
+ 2,
+ 3)))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage(
+ "numberOfWriteResults is not same across TableBucketWriteResults for table %s, got %s and %s.",
+ tableId, 2, 3);
+ verifyNoLakeSnapshot(tablePath);
+ }
+
@Test
void testCommitNonPartitionedTable() throws Exception {
TablePath tablePath1 = TablePath.of("fluss", "test1");
@@ -299,7 +347,7 @@ class TieringCommitOperatorTest extends FlinkTestBase {
void testTableCommitWhenFlussMissingLakeSnapshot() throws Exception {
TablePath tablePath = TablePath.of("fluss", "test_commit_when_fluss_missing_lake_snapshot");
long tableId = createTable(tablePath, DEFAULT_PK_TABLE_DESCRIPTOR);
- int numberOfWriteResults = 3;
+ int numberOfWriteResults = 2;
Map<TableBucket, Long> expectedLogEndOffsets = new HashMap<>();
for (int bucket = 0; bucket < 3; bucket++) {
@@ -320,7 +368,7 @@ class TieringCommitOperatorTest extends FlinkTestBase {
new TestingLakeTieringFactory(testingLakeCommitter));
committerOperator.open();
- for (int bucket = 0; bucket < 3; bucket++) {
+ for (int bucket = 0; bucket < numberOfWriteResults; bucket++) {
TableBucket tableBucket = new TableBucket(tableId, bucket);
committerOperator.processElement(
createTableBucketWriteResultStreamRecord(
@@ -340,19 +388,33 @@ class TieringCommitOperatorTest extends FlinkTestBase {
tablePath,
tableId,
mockMissingCommittedLakeSnapshot));
+ assertThat(output).isEmpty();
+ // Retry with a different count and arrival order to verify the failed round was cleared.
+ numberOfWriteResults = 3;
expectedLogEndOffsets = new HashMap<>();
- for (int bucket = 0; bucket < 3; bucket++) {
+ for (int bucket : new int[] {2, 0, 1}) {
TableBucket tableBucket = new TableBucket(tableId, bucket);
long offset = bucket * bucket;
long timestamp = bucket * bucket;
committerOperator.processElement(
createTableBucketWriteResultStreamRecord(
- tablePath, tableBucket, 3, offset, timestamp, numberOfWriteResults));
+ tablePath,
+ tableBucket,
+ bucket,
+ offset,
+ timestamp,
+ numberOfWriteResults));
expectedLogEndOffsets.put(tableBucket, offset);
}
verifyLakeSnapshot(tablePath, tableId, 1, expectedLogEndOffsets);
+ assertThat(output)
+ .singleElement()
+ .satisfies(
+ message ->
+ assertThat(message.committable().writeResults())
+ .containsExactly(2, 0, 1));
}
@Test
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorSchedulingTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorSchedulingTest.java
index 2bc15b7a3..2a4b24145 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorSchedulingTest.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorSchedulingTest.java
@@ -52,9 +52,43 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
-/** Tests for empty-table continuation in {@link TieringSourceEnumerator}. */
+/** Tests for split assignment and table scheduling in {@link TieringSourceEnumerator}. */
class TieringSourceEnumeratorSchedulingTest {
+ @Test
+ void testPendingSplitsAssignedInOrder() throws Exception {
+ try (Fixture fixture = new Fixture()) {
+ List<TieringSplit> splits = new ArrayList<>();
+ for (int bucket = 0; bucket < 4; bucket++) {
+ splits.add(
+ new TieringLogSplit(
+ TablePath.of("db", "table"),
+ new TableBucket(1, bucket),
+ null,
+ 0,
+ 1));
+ }
+ fixture.enumerator.addSplitsBack(splits.subList(0, 2), 0);
+ // Returned splits join the tail of the remaining queue.
+ fixture.enumerator.addSplitsBack(splits.subList(2, splits.size()), 0);
+ for (int i = 2; i < splits.size(); i++) {
+ fixture.requestTable();
+ }
+
+ List<TieringSplit> assignedSplits =
+ fixture.context.getSplitsAssignmentSequence().stream()
+ .flatMap(assignment -> assignment.assignment().get(0).stream())
+ .collect(Collectors.toList());
+ assertThat(assignedSplits).containsExactlyElementsOf(splits);
+ assertThat(fixture.claims()).isEmpty();
+
+ // Once drained, another request polls for a new table without reassigning old splits.
+ fixture.requestTable();
+ assertThat(fixture.claims()).hasSize(1);
+ assertThat(fixture.context.getSplitsAssignmentSequence()).hasSize(splits.size());
+ }
+ }
+
@Test
void testEmptyTablesShareContinuationAndAdvance() throws Throwable {
try (Fixture fixture = new Fixture()) {
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java
index 8d1f62eb3..c91a960e9 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java
@@ -32,6 +32,7 @@ import org.apache.fluss.flink.tiering.source.split.TieringSplitGenerator;
import org.apache.fluss.lake.writer.LakeTieringFactory;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableChange;
+import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.rpc.messages.CommitLakeTableSnapshotRequest;
@@ -599,13 +600,17 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
@Test
void testHandleReaderFailOver() throws Throwable {
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder(DEFAULT_LOG_TABLE_DESCRIPTOR)
+ .distributedBy(4, "id")
+ .build();
TablePath tablePath1 = TablePath.of(DEFAULT_DB, "tiering-failover-test-log-table1");
- createTable(tablePath1, DEFAULT_LOG_TABLE_DESCRIPTOR);
- appendRow(tablePath1, DEFAULT_LOG_TABLE_DESCRIPTOR, 0, 10);
+ createTable(tablePath1, tableDescriptor);
+ appendRow(tablePath1, tableDescriptor, 0, 10);
TablePath tablePath2 = TablePath.of(DEFAULT_DB, "tiering-failover-test-log-table2");
- createTable(tablePath2, DEFAULT_LOG_TABLE_DESCRIPTOR);
- appendRow(tablePath2, DEFAULT_LOG_TABLE_DESCRIPTOR, 0, 10);
+ createTable(tablePath2, tableDescriptor);
+ appendRow(tablePath2, tableDescriptor, 0, 10);
try (FlussMockSplitEnumeratorContext<TieringSplit> context =
new FlussMockSplitEnumeratorContext<>(3)) {
@@ -617,8 +622,13 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
// register readers and handle split requests for attempt 0
registerReaderAndHandleSplitRequests(context, enumerator, 3, 0);
- // should get one tiering split, and the split is for tablePath1
+ // Assign three of the four splits, leaving one pending when failover starts.
verifyTieringSplitAssignment(context, 3, tablePath1);
+ assertThat(context.getSplitsAssignmentSequence())
+ .flatExtracting(assignment -> assignment.assignment().values())
+ .flatExtracting(splits -> splits)
+ .extracting(TieringSplit::getNumberOfSplits)
+ .containsOnly(4);
// clean assignment
context.getSplitsAssignmentSequence().clear();
@@ -627,8 +637,7 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
// register readers and handle split requests (attempt 1)
registerReaderAndHandleSplitRequests(context, enumerator, 3, 1);
- // now, should get another one tiering split, the split is for tablePath2 since all
- // pending split for tablePath1 is clear
+ // Only tablePath2 splits are assigned; the pending tablePath1 split was cleared.
verifyTieringSplitAssignment(context, 3, tablePath2);
// clean assignment
@@ -642,8 +651,7 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
// tablePath2 is clear and there still one sub-task is not registered
verifyTieringSplitAssignment(context, 0, tablePath2);
- // register reader 2 again, should get tiering split for table1 since the failover is
- // finished, and reader2 request tiering split
+ // Once reader 2 restarts, failover completes and tablePath1 splits can be assigned.
registerSingleReaderAndHandleSplitRequests(context, enumerator, 2, 2);
verifyTieringSplitAssignment(context, 3, tablePath1);
}