This is an automated email from the ASF dual-hosted git repository.
loserwang1024 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 4b678829b [FLINK-40740][pipeline-connector/fluss] Prevent duplicate
split initialization during discovery (#4555)
4b678829b is described below
commit 4b678829ba2c0f17c68ff015113844f05945d702
Author: Xiaobing Fang <[email protected]>
AuthorDate: Tue Sep 29 14:19:09 2026 +0800
[FLINK-40740][pipeline-connector/fluss] Prevent duplicate split
initialization during discovery (#4555)
---
.../source/enumerator/FlussSourceEnumerator.java | 42 +-
.../enumerator/FlussSourceEnumeratorTest.java | 437 +++++++++++++++++++++
2 files changed, 459 insertions(+), 20 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumerator.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumerator.java
index efe8945ee..0d30fcb61 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumerator.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumerator.java
@@ -82,8 +82,9 @@ import java.util.stream.Collectors;
* <ol>
* <li>{@link #getSubscribedTableBuckets()} — discovers subscribed tables
and enumerates all
* table-buckets including partitions (async).
- * <li>{@link #checkTableBucketChanges} — compares discovered table-buckets
with already-assigned
- * ones and triggers split creation for new table-buckets (callback).
+ * <li>{@link #checkTableBucketChanges} — compares discovered table-buckets
with assigned,
+ * pending, and initializing ones and triggers split creation for new
table-buckets
+ * (callback).
* <li>{@link #initPendingBucketSplits} — resolves starting offsets and
creates splits for new
* table-buckets (async).
* <li>{@link #handleTableBucketChanges} — marks physical table paths as
assigned and distributes
@@ -120,7 +121,7 @@ public class FlussSourceEnumerator
private final Map<TablePath, Long> pendingRemovalRequests;
private final Map<TablePath, Set<Integer>> removalAcknowledgements;
private final Map<TableBucket, TablePath> pendingRemovalBuckets;
- private final Set<TablePath> initializingTablePaths;
+ private final Set<PhysicalTablePath> initializingPhysicalTablePaths;
private final Map<TablePath, Long> removalFences;
private final Map<String, FlussSplitBase> fencedFreshSplits;
private List<TableBucketInfo> lastDiscoveredTableBuckets;
@@ -158,7 +159,7 @@ public class FlussSourceEnumerator
this.pendingRemovalRequests = new HashMap<>();
this.removalAcknowledgements = new HashMap<>();
this.pendingRemovalBuckets = new HashMap<>();
- this.initializingTablePaths = new HashSet<>();
+ this.initializingPhysicalTablePaths = new HashSet<>();
this.removalFences = new HashMap<>();
this.fencedFreshSplits = new HashMap<>();
this.lastDiscoveredTableBuckets = Collections.emptyList();
@@ -287,8 +288,8 @@ public class FlussSourceEnumerator
//
-------------------------------------------------------------------------
/**
- * Compares the discovered table-buckets against assigned and pending
{@link PhysicalTablePath}s
- * and triggers split creation for newly discovered table-buckets.
+ * Compares the discovered table-buckets against assigned, pending, and
initializing {@link
+ * PhysicalTablePath}s and triggers split creation for newly discovered
table-buckets.
*/
private void checkTableBucketChanges(DiscoveryResult discoveryResult,
Throwable error) {
if (error != null) {
@@ -308,18 +309,17 @@ public class FlussSourceEnumerator
private void initializeNewTableBuckets(List<TableBucketInfo> allBuckets) {
- Set<PhysicalTablePath> assignedOrPendingPhysicalTablePaths =
- new HashSet<>(assignedPhysicalTablePaths);
+ Set<PhysicalTablePath> knownPhysicalTablePaths = new
HashSet<>(assignedPhysicalTablePaths);
+ knownPhysicalTablePaths.addAll(initializingPhysicalTablePaths);
pendingPartitionSplitAssignment.values().stream()
.flatMap(Set::stream)
.map(FlussSplitBase::getPhysicalTablePath)
- .forEach(assignedOrPendingPhysicalTablePaths::add);
+ .forEach(knownPhysicalTablePaths::add);
List<TableBucketInfo> newBuckets = new ArrayList<>();
for (TableBucketInfo info : allBuckets) {
- if
(!initializingTablePaths.contains(info.physicalTablePath.getTablePath())
- &&
!pendingRemovalTablePaths.contains(info.physicalTablePath.getTablePath())
- &&
!assignedOrPendingPhysicalTablePaths.contains(info.physicalTablePath)) {
+ if
(!pendingRemovalTablePaths.contains(info.physicalTablePath.getTablePath())
+ &&
!knownPhysicalTablePaths.contains(info.physicalTablePath)) {
newBuckets.add(info);
}
}
@@ -330,11 +330,9 @@ public class FlussSourceEnumerator
}
LOG.info("Discovered {} new table-bucket(s) to initialize.",
newBuckets.size());
- Set<TablePath> initializingPaths =
- newBuckets.stream()
- .map(info -> info.physicalTablePath.getTablePath())
- .collect(Collectors.toSet());
- initializingTablePaths.addAll(initializingPaths);
+ Set<PhysicalTablePath> initializingPaths =
+ newBuckets.stream().map(info ->
info.physicalTablePath).collect(Collectors.toSet());
+ initializingPhysicalTablePaths.addAll(initializingPaths);
context.callAsync(
() -> initPendingBucketSplits(newBuckets),
(splits, error) -> handleTableBucketChanges(splits, error,
initializingPaths));
@@ -388,7 +386,8 @@ public class FlussSourceEnumerator
}
Set<TablePath> clearedTablePaths = new HashSet<>();
for (TablePath tablePath : pendingRemovalTablePaths) {
- if (!initializingTablePaths.contains(tablePath)
+ if (initializingPhysicalTablePaths.stream()
+ .noneMatch(path ->
path.getTablePath().equals(tablePath))
&& removalAcknowledgements
.getOrDefault(tablePath, Collections.emptySet())
.containsAll(expectedReaders)) {
@@ -634,11 +633,14 @@ public class FlussSourceEnumerator
* distributes the splits to registered readers.
*/
private void handleTableBucketChanges(
- List<FlussSplitBase> newSplits, Throwable error, Set<TablePath>
initializingPaths) {
- initializingTablePaths.removeAll(initializingPaths);
+ List<FlussSplitBase> newSplits,
+ Throwable error,
+ Set<PhysicalTablePath> initializingPaths) {
+ initializingPhysicalTablePaths.removeAll(initializingPaths);
if (error != null) {
boolean hasActiveInitializingPath =
initializingPaths.stream()
+ .map(PhysicalTablePath::getTablePath)
.anyMatch(
tablePath ->
subscribedTablePaths.contains(tablePath)
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumeratorTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumeratorTest.java
index 3fe9653a4..bb0731193 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumeratorTest.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumeratorTest.java
@@ -55,6 +55,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import java.time.Duration;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -475,6 +476,168 @@ class FlussSourceEnumeratorTest {
}
}
+ /**
+ * Tests that running several periodic discovery rounds before the async
split initialization
+ * triggered by the first round completes does not queue duplicate
initializations for the same
+ * table-bucket.
+ */
+ @Test
+ void testMultiplePeriodicDiscoveryRoundsBeforeInitDoNotDuplicateSplits()
throws Throwable {
+ String tableName = "multi_bucket_dup";
+ createPkTable(tableName, 3);
+
+ FlussDefaultDiscoverer discoverer = new FlussDefaultDiscoverer();
+ String pattern = fqnRegex(DATABASE_NAME, tableName);
+
+ AtomicInteger offsetInitializationCount = new AtomicInteger();
+ OffsetsInitializer earliestOffsetsInitializer =
OffsetsInitializer.earliest();
+ OffsetsInitializer countingOffsetsInitializer =
+ (partitionName, bucketIds, retriever) -> {
+ offsetInitializationCount.incrementAndGet();
+ return earliestOffsetsInitializer.getBucketOffsets(
+ partitionName, bucketIds, retriever);
+ };
+
+ try (MockSplitEnumeratorContext<FlussSplitBase> context =
+ new MockSplitEnumeratorContext<>(NUM_READERS)) {
+ FlussSourceEnumerator enumerator =
+ newEnumerator(
+ context,
+ discoverer,
+ pattern,
+ countingOffsetsInitializer,
+ DISCOVERY_INTERVAL_MS);
+ try {
+ enumerator.start();
+ registerAllReaders(context, enumerator);
+
+ // Simulates a slow/backed-up async initialization: several
discovery rounds run
+ // before any of the resulting one-time (init) callables are
drained.
+ context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX);
+ context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX);
+ context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX);
+
+ int queuedInitCount = context.getOneTimeCallables().size();
+
+ while (!context.getOneTimeCallables().isEmpty()) {
+ context.runNextOneTimeCallable();
+ }
+
+ assertThat(assignedBucketIds(context, tableName))
+ .containsExactlyInAnyOrder(0, 1, 2);
+ assertThat(queuedInitCount)
+ .as(
+ "Each already-discovered table-bucket must
only be queued for "
+ + "initialization once, even across
repeated discovery "
+ + "rounds")
+ .isEqualTo(1);
+ assertThat(offsetInitializationCount.get())
+ .as("The offsets initializer must only be invoked once
for the table")
+ .isEqualTo(1);
+
+ context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX);
+ assertThat(context.getOneTimeCallables())
+ .as("Nothing new to discover once every bucket is
already assigned")
+ .isEmpty();
+ } finally {
+ enumerator.close();
+ }
+ }
+ }
+
+ /**
+ * Tests that when no readers are registered yet, a completed table-bucket
initialization lands
+ * in the pending assignment map instead of being lost or re-queued, and
that once readers
+ * register, all pending splits are assigned exactly once.
+ */
+ @Test
+ void testPendingAssignmentSurvivesDiscoveryUntilReadersRegister() throws
Throwable {
+ String tableName = "pending_no_reader";
+ createPkTable(tableName, 3);
+
+ FlussDefaultDiscoverer discoverer = new FlussDefaultDiscoverer();
+ String pattern = fqnRegex(DATABASE_NAME, tableName);
+
+ try (MockSplitEnumeratorContext<FlussSplitBase> context =
+ new MockSplitEnumeratorContext<>(NUM_READERS)) {
+ FlussSourceEnumerator enumerator = newEnumerator(context,
discoverer, pattern);
+ try {
+ enumerator.start();
+
+ // No readers registered yet: initialization completes into
the pending
+ // assignment map instead of being assigned.
+ runDiscoveryCycle(context);
+ assertThat(context.getSplitsAssignmentSequence()).isEmpty();
+
+ // A subsequent discovery round must not re-discover or
re-queue the same
+ // table-buckets while they are only pending (not yet
assigned).
+ context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX);
+ assertThat(context.getOneTimeCallables()).isEmpty();
+
+ registerAllReaders(context, enumerator);
+
+ assertThat(assignedBucketIds(context, tableName))
+ .containsExactlyInAnyOrder(0, 1, 2);
+ } finally {
+ enumerator.close();
+ }
+ }
+ }
+
+ /**
+ * Tests that once one partition's initialization completes, a subsequent
discovery round does
+ * not re-queue initialization for another partition whose initialization
is still in flight.
+ */
+ @Test
+ void
testDiscoveryDoesNotDropStillInitializingPartitionAfterAnotherCompletes()
+ throws Throwable {
+ String tableName = "two_partition_inflight";
+ createPartitionedPkTable(tableName, 3);
+ insertPartitionRow(tableName, 1, "p1", "a");
+
+ FlussDefaultDiscoverer discoverer = new FlussDefaultDiscoverer();
+ String pattern = fqnRegex(DATABASE_NAME, tableName);
+
+ try (MockSplitEnumeratorContext<FlussSplitBase> context =
+ new MockSplitEnumeratorContext<>(NUM_READERS)) {
+ FlussSourceEnumerator enumerator = newEnumerator(context,
discoverer, pattern);
+ try {
+ enumerator.start();
+ registerAllReaders(context, enumerator);
+
+ // Round 1: discovers partition p1 and queues its
initialization, without
+ // draining it.
+ context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX);
+ assertThat(context.getOneTimeCallables()).hasSize(1);
+
+ // Partition p2 appears before p1's initialization has
completed.
+ insertPartitionRow(tableName, 2, "p2", "b");
+
+ // Round 2: discovers p2 and queues its initialization; p1 is
still in flight.
+ context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX);
+ assertThat(context.getOneTimeCallables()).hasSize(2);
+
+ context.runNextOneTimeCallable();
+ assertThat(assignedSplitIds(context, tableName))
+ .containsExactlyInAnyOrder(expectedSplitIds(tableName,
"p1"));
+ assertThat(context.getOneTimeCallables()).hasSize(1);
+
+ // Round 3: p1 is now assigned, p2 is still in flight (its
callable has not been
+ // drained yet). This round must not re-queue p2's
initialization.
+ context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX);
+ assertThat(context.getOneTimeCallables())
+ .as("A still in-flight partition must not be re-queued
for initialization")
+ .hasSize(1);
+
+ context.runNextOneTimeCallable();
+ assertThat(assignedSplitIds(context, tableName))
+ .containsExactlyInAnyOrder(expectedSplitIds(tableName,
"p1", "p2"));
+ } finally {
+ enumerator.close();
+ }
+ }
+ }
+
@Test
void testDiscoveryFailsWhenTableDiscoveryFails() throws Throwable {
RuntimeException discoveryFailure =
@@ -552,6 +715,89 @@ class FlussSourceEnumeratorTest {
}
}
+ /**
+ * Tests that after an offset-initialization failure, restarting the
enumerator from a
+ * checkpoint taken at that point fully rediscovers and initializes every
table-bucket exactly
+ * once. The failed instance is discarded rather than retried in place:
{@link
+ * MockSplitEnumeratorContext}'s stored worker error is not cleared once
thrown, so it cannot be
+ * reused to observe a same-instance retry.
+ */
+ @Test
+ void
testDiscoveryFullyRediscoversAfterOffsetInitializationFailureOnRestart() throws
Throwable {
+ String tableName = "offset_failure_restart";
+ createPkTable(tableName, 3);
+
+ RuntimeException offsetFailure =
+ new RuntimeException("Injected offset initialization failure");
+ OffsetsInitializer failingOffsetsInitializer =
+ (partitionName, bucketIds, retriever) -> {
+ throw offsetFailure;
+ };
+
+ FlussSourceEnumState checkpoint;
+ try (MockSplitEnumeratorContext<FlussSplitBase> context =
+ new MockSplitEnumeratorContext<>(NUM_READERS)) {
+ FlussSourceEnumerator enumerator =
+ newEnumerator(
+ context,
+ new FlussDefaultDiscoverer(),
+ fqnRegex(DATABASE_NAME, tableName),
+ failingOffsetsInitializer,
+ DISCOVERY_INTERVAL_MS);
+ try {
+ enumerator.start();
+ registerAllReaders(context, enumerator);
+
+ context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX);
+ assertThatThrownBy(context::runNextOneTimeCallable)
+ .isInstanceOf(FlinkRuntimeException.class)
+ .hasMessage("Failed to initialize splits for new
table-buckets.")
+ .hasRootCauseMessage("Injected offset initialization
failure");
+ assertThat(context.getSplitsAssignmentSequence()).isEmpty();
+
+ checkpoint = enumerator.snapshotState(1L);
+
assertThat(checkpoint.getAssignedPhysicalTablePaths()).isEmpty();
+ assertThat(checkpoint.getRemainingSplits()).isEmpty();
+ } finally {
+ enumerator.close();
+ }
+ }
+
+ FlussSourceEnumStateSerializer serializer = new
FlussSourceEnumStateSerializer();
+ FlussSourceEnumState restoredCheckpoint =
+ serializer.deserialize(serializer.getVersion(),
serializer.serialize(checkpoint));
+
+ try (MockSplitEnumeratorContext<FlussSplitBase> restoredContext =
+ new MockSplitEnumeratorContext<>(NUM_READERS)) {
+ org.apache.fluss.config.Configuration flussConfig =
+ FLUSS_CLUSTER_EXTENSION.getClientConfig();
+ FlussSourceEnumerator restoredEnumerator =
+ new FlussSourceEnumerator(
+ restoredContext,
+ new FlussDefaultDiscoverer(),
+ flussConfig,
+ buildSourceConfig(flussConfig,
fqnRegex(DATABASE_NAME, tableName)),
+ OffsetsInitializer.earliest(),
+ DISCOVERY_INTERVAL_MS,
+ restoredCheckpoint,
+ LeaseContext.fromConf(
+ new
org.apache.flink.configuration.Configuration()));
+ try {
+ restoredEnumerator.start();
+ registerAllReaders(restoredContext, restoredEnumerator);
+
+ runDiscoveryCycle(restoredContext);
+ assertThat(assignedBucketIds(restoredContext, tableName))
+ .containsExactlyInAnyOrder(0, 1, 2);
+
+ restoredContext.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX);
+ assertThat(restoredContext.getOneTimeCallables()).isEmpty();
+ } finally {
+ restoredEnumerator.close();
+ }
+ }
+ }
+
@Test
void testDiscoveryFailsWhenOffsetInitializationIsIncomplete() throws
Throwable {
String tableName = "incomplete_offsets";
@@ -805,6 +1051,74 @@ class FlussSourceEnumeratorTest {
}
}
+ /**
+ * Tests that a table-bucket whose initialization is still in flight at
snapshot time (i.e. not
+ * yet reflected in {@link FlussSourceEnumState}) is fully re-discovered
and initialized exactly
+ * once after restoring from that snapshot.
+ */
+ @Test
+ void
testSnapshotDuringInFlightInitializationAllowsFullRediscoveryAfterRestore()
+ throws Throwable {
+ String tableName = "inflight_snapshot_restore";
+ createPkTable(tableName, 3);
+
+ FlussSourceEnumState checkpoint;
+ try (MockSplitEnumeratorContext<FlussSplitBase> context =
+ new MockSplitEnumeratorContext<>(NUM_READERS)) {
+ FlussSourceEnumerator enumerator =
+ newEnumerator(
+ context,
+ new FlussDefaultDiscoverer(),
+ fqnRegex(DATABASE_NAME, tableName));
+ try {
+ enumerator.start();
+ registerAllReaders(context, enumerator);
+
+ // Discovered but not drained: tracked as in-flight, not yet
pending or assigned.
+ context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX);
+ assertThat(context.getOneTimeCallables()).hasSize(1);
+
+ checkpoint = enumerator.snapshotState(1L);
+
assertThat(checkpoint.getAssignedPhysicalTablePaths()).isEmpty();
+ assertThat(checkpoint.getRemainingSplits()).isEmpty();
+ } finally {
+ enumerator.close();
+ }
+ }
+
+ FlussSourceEnumStateSerializer serializer = new
FlussSourceEnumStateSerializer();
+ FlussSourceEnumState restoredCheckpoint =
+ serializer.deserialize(serializer.getVersion(),
serializer.serialize(checkpoint));
+
+ try (MockSplitEnumeratorContext<FlussSplitBase> restoredContext =
+ new MockSplitEnumeratorContext<>(NUM_READERS)) {
+ org.apache.fluss.config.Configuration flussConfig =
+ FLUSS_CLUSTER_EXTENSION.getClientConfig();
+ FlussSourceEnumerator restoredEnumerator =
+ new FlussSourceEnumerator(
+ restoredContext,
+ new FlussDefaultDiscoverer(),
+ flussConfig,
+ buildSourceConfig(flussConfig,
fqnRegex(DATABASE_NAME, tableName)),
+ OffsetsInitializer.earliest(),
+ DISCOVERY_INTERVAL_MS,
+ restoredCheckpoint,
+ LeaseContext.fromConf(
+ new
org.apache.flink.configuration.Configuration()));
+ try {
+ restoredEnumerator.start();
+ registerAllReaders(restoredContext, restoredEnumerator);
+
+ runDiscoveryCycle(restoredContext);
+
+ assertThat(assignedBucketIds(restoredContext, tableName))
+ .containsExactlyInAnyOrder(0, 1, 2);
+ } finally {
+ restoredEnumerator.close();
+ }
+ }
+ }
+
// =====================================================================
// FlussTableSubscriber tests — subscription-table driven add/remove
// =====================================================================
@@ -1056,6 +1370,68 @@ class FlussSourceEnumeratorTest {
}
}
+ @Test
+ void testResubscriptionWaitsForAllInitializingPartitions() throws
Throwable {
+ String subscriptionTable = "sub_partition_inflight";
+ String targetTable = "tgt_partition_inflight";
+ TablePath tablePath = TablePath.of(DATABASE_NAME, targetTable);
+ createSubscriptionTable(subscriptionTable);
+ createPartitionedPkTable(targetTable, 3);
+ insertPartitionRow(targetTable, 1, "p1", "a");
+ insertSubscription(subscriptionTable, targetTable);
+
+ FlussSubscriberTableDiscoverer subscriber =
+ new FlussSubscriberTableDiscoverer(DATABASE_NAME + "." +
subscriptionTable, 100);
+ try (MockSplitEnumeratorContext<FlussSplitBase> context =
+ new MockSplitEnumeratorContext<>(NUM_READERS)) {
+ FlussSourceEnumerator enumerator = newEnumerator(context,
subscriber, null);
+ try {
+ enumerator.start();
+ registerAllReaders(context, enumerator);
+ context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX);
+ insertPartitionRow(targetTable, 2, "p2", "b");
+ context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX);
+ assertThat(context.getOneTimeCallables()).hasSize(2);
+
+ tBatchEnv
+ .executeSql(
+ String.format(
+ "DELETE FROM %s WHERE table_name =
'%s.%s'",
+ subscriptionTable, DATABASE_NAME,
targetTable))
+ .await();
+ context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX);
+ long requestId =
+ latestSubscriptionEvent(context, 0)
+ .getPendingRemovalRequests()
+ .get(tablePath);
+ acknowledgeRemoval(enumerator, tablePath, requestId);
+ insertSubscription(subscriptionTable, targetTable);
+ context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX);
+ assertThat(context.getOneTimeCallables()).hasSize(2);
+
+ // Reader acknowledgements cannot release the other
partition's in-flight work.
+ context.runNextOneTimeCallable();
+
assertThat(enumerator.snapshotState(1L).getPendingRemovalTablePaths())
+ .containsExactly(tablePath);
+ assertThat(context.getSplitsAssignmentSequence()).isEmpty();
+ assertThat(context.getOneTimeCallables()).hasSize(1);
+
+ context.runNextOneTimeCallable();
+
assertThat(enumerator.snapshotState(2L).getPendingRemovalTablePaths()).isEmpty();
+ assertThat(context.getSplitsAssignmentSequence()).isEmpty();
+ assertThat(context.getOneTimeCallables()).hasSize(1);
+
+ context.runNextOneTimeCallable();
+ assertThat(assignedSplitIds(context, targetTable))
+
.containsExactlyInAnyOrder(expectedSplitIds(targetTable, "p1", "p2"));
+ context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX);
+ assertThat(context.getOneTimeCallables()).isEmpty();
+ } finally {
+ enumerator.close();
+ }
+ }
+ }
+
@Test
void testLateInitializationDoesNotReviveDeletedSubscription() throws
Throwable {
String subscriptionTable = "sub_late_init";
@@ -1506,6 +1882,46 @@ class FlussSourceEnumeratorTest {
.collect(Collectors.toSet());
}
+ /** Returns the bucket ids assigned so far for the given table, across all
assignments. */
+ private static List<Integer> assignedBucketIds(
+ MockSplitEnumeratorContext<FlussSplitBase> context, String
tableName) {
+ return context.getSplitsAssignmentSequence().stream()
+ .flatMap(assignment ->
assignment.assignment().values().stream())
+ .flatMap(List::stream)
+ .filter(split ->
split.getPhysicalTablePath().getTableName().equals(tableName))
+ .map(split -> split.getTableBucket().getBucket())
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Returns the full split id of every split assigned so far for the given
table, across all
+ * assignments. A {@link List} (not a {@link Set}) is returned so
duplicate splits are not
+ * silently hidden.
+ */
+ private static List<String> assignedSplitIds(
+ MockSplitEnumeratorContext<FlussSplitBase> context, String
tableName) {
+ return context.getSplitsAssignmentSequence().stream()
+ .flatMap(assignment ->
assignment.assignment().values().stream())
+ .flatMap(List::stream)
+ .filter(split ->
split.getPhysicalTablePath().getTableName().equals(tableName))
+ .map(FlussSplitBase::splitId)
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Builds the expected {@link FlussSplitBase#splitId()} values for buckets
0..2 of the given
+ * table and partitions.
+ */
+ private static String[] expectedSplitIds(String tableName, String...
partitionNames) {
+ List<String> ids = new ArrayList<>();
+ for (String partitionName : partitionNames) {
+ for (int bucket = 0; bucket < 3; bucket++) {
+ ids.add(DATABASE_NAME + "." + tableName + "." + partitionName
+ "." + bucket);
+ }
+ }
+ return ids.toArray(new String[0]);
+ }
+
/** Returns the set of table names in the most recent assignment only. */
private static Set<String> latestAssignmentTableNames(
MockSplitEnumeratorContext<FlussSplitBase> context) {
@@ -1560,6 +1976,27 @@ class FlussSourceEnumeratorTest {
}
}
+ private void createPartitionedPkTable(String tableName, int numBuckets)
throws Exception {
+ tBatchEnv
+ .executeSql(
+ String.format(
+ "CREATE TABLE %s (id INT, ds STRING, val
STRING, "
+ + "PRIMARY KEY (id, ds) NOT ENFORCED)
PARTITIONED BY (ds) "
+ + "WITH ('bucket.num' = '%d')",
+ tableName, numBuckets))
+ .await();
+ }
+
+ private void insertPartitionRow(String tableName, int id, String
partitionValue, String val)
+ throws Exception {
+ tBatchEnv
+ .executeSql(
+ String.format(
+ "INSERT INTO %s VALUES (%d, '%s', '%s')",
+ tableName, id, partitionValue, val))
+ .await();
+ }
+
private void createSubscriptionTable(String tableName) throws Exception {
tBatchEnv
.executeSql(
