[connector] Fix partition state update and filter non-existent partitions on restore #359

codope · 2025-02-08T07:34:37Z

Purpose

Linked issue: close #348

This PR fixes an issue where removed partitions were still retained in the source reader’s state. Upon recovery, the source attempted to re-subscribe to non-existent partitions, leading to a PartitionNotExistException.

Changes

Mark Finished Splits on Partition Removal:
Active bounded splits are now marked as finished when a partition removal event is received. This prevents removed partitions from being restored.
Remove Pending Snapshot Splits and Unsubscribe Buckets:
Pending snapshot splits and subscribed log buckets associated with removed partitions are removed and unsubscribed respectively.
Filter Non-existent Partitions During Restore:
The subscribeLog method now catches PartitionNotExistException and logs a warning instead of failing, filtering out any non-existent partitions on restore.

Tests

Added FlinkTableSourceFailOverITCase integration test.

API and Format

None

Documentation

NA

CLAassistant · 2025-02-08T07:34:43Z

All committers have signed the CLA.

...k/src/test/java/com/alibaba/fluss/connector/flink/source/FlinkTableSourceFailOverITCase.java

- Mark active bounded splits as finished on partition removal. - Remove pending snapshot splits and unsubscribe dropped log buckets. - Catch PartitionNotExistException during subscribeLog to log a warning.

wuchong · 2025-02-20T10:08:29Z

...nk/src/main/java/com/alibaba/fluss/connector/flink/source/reader/FlinkSourceSplitReader.java

+                // Traverse the exception chain to check for PartitionNotExistException.
+                Throwable cause = e;
+                boolean partitionNotExist = false;
+                while (cause != null) {


nit: we can replace the while loop using

boolean partitionNotExist = ExceptionUtils.findThrowable(e, PartitionNotExistException.class).isPresent();

wuchong · 2025-02-20T11:04:49Z

...nk/src/main/java/com/alibaba/fluss/connector/flink/source/reader/FlinkSourceSplitReader.java

+                } catch (IOException e) {
+                    LOG.warn(
+                            "Failed to close current bounded split for removed partition {}.",
+                            currentBucket.getPartitionId(),


nit: we can also use removedPartitions.get(currentBucket.getPartitionId()) to print the partition name rather than id.

wuchong · 2025-02-20T17:25:38Z

...nk/src/main/java/com/alibaba/fluss/connector/flink/source/reader/FlinkSourceSplitReader.java

+                    LOG.warn(
+                            "Partition {} does not exist when subscribing to log for split {}. Skipping subscription.",
+                            partitionId,
+                            split.splitId());


should return to avoid unnecessary subscribe the partition.

wuchong · 2025-02-20T17:30:13Z

Thanks @codope for the contribution! The fixing looks good to me. I pushed a commit to improve the IT case so that:

avoid Thread.sleep(..) as it will lead tests unstable if CPU is high.
Verify with data instead of verifying with no exception.
the new IT case can also reproduce the original exception mentioned in issue description.

Could you help to review the changes @luoyuxia ?

luoyuxia

Thanks @codope @wuchong Overall LGTM. Just left minor comments.

luoyuxia · 2025-02-21T06:12:57Z

...nk/src/main/java/com/alibaba/fluss/connector/flink/source/reader/FlinkSourceSplitReader.java

@@ -262,17 +287,42 @@ public Set<TableBucket> removePartitions(Map<Long, String> removedPartitions) {
        // todo, may consider to close the current snapshot reader if


The todo comment can be removed now since in this pr, we wil close currentBoundedSplit

luoyuxia · 2025-02-21T06:14:41Z

...nk/src/main/java/com/alibaba/fluss/connector/flink/source/reader/FlinkSourceSplitReader.java

@@ -145,6 +149,12 @@ public RecordsWithSplitIds<RecordAndPos> fetch() throws IOException {
                                new HashSet<>(emptyLogSplits), flinkSourceReaderMetrics);
                emptyLogSplits.clear();
                return records;
+            } else if (!removedSplits.isEmpty()) {


I think we can check !removedSplits.isEmpty() at the beginning so that the flink enumerate can remove the state of removed splits as soon as possible.

Yes, we can move the code block at the beginning of fetch() method, because the currentBoundedSplitReader may still work in progress.

luoyuxia · 2025-02-21T06:20:24Z

...nk/src/main/java/com/alibaba/fluss/connector/flink/source/reader/FlinkSourceSplitReader.java

+            try {
+                if (partitionId != null) {
+                    // Try to subscribe using the partition id.
+                    logScanner.subscribe(partitionId, bucket, startingOffset);


nit:
I think we can limit the try catch scope to only this line since only subscribe a partition, the partitionNotExist will be thrown.
I'd like suggest to be:

if (partitionId != null) { // Try to subscribe using the partition id. try { logScanner.subscribe(partitionId, bucket, startingOffset); } catch (Exception e) { // the PartitionNotExistException may still happens when partition is removed // but Flink source reader failover before aware of it // Traverse the exception chain to check for PartitionNotExistException. boolean partitionNotExist = ExceptionUtils.findThrowable(e, PartitionNotExistException.class) .isPresent(); if (partitionNotExist) { // mark the not exist partition to be removed removedSplits.add(split.splitId()); LOG.warn( "Partition {} does not exist when subscribing to log for split {}. Skipping subscription.", partitionId, split.splitId()); return; } } } else { // If no partition id, subscribe by bucket only. logScanner.subscribe(bucket, startingOffset); }

luoyuxia · 2025-02-21T06:23:39Z

...nk/src/main/java/com/alibaba/fluss/connector/flink/source/reader/FlinkSourceSplitReader.java

+                        ExceptionUtils.findThrowable(e, PartitionNotExistException.class)
+                                .isPresent();
+                if (partitionNotExist) {
+                    LOG.warn(


We also should mark split as removed since the partition is removed and so that flink enumerator can remove this split state.

wuchong · 2025-02-21T07:38:00Z

@codope do you have time to address @luoyuxia 's comments? If you are busy, I can help with that.

codope commented Feb 8, 2025

View reviewed changes

...k/src/test/java/com/alibaba/fluss/connector/flink/source/FlinkTableSourceFailOverITCase.java Outdated Show resolved Hide resolved

codope mentioned this pull request Feb 8, 2025

PartitionNotExistException will be thrown when restore from checkpoint #348

Open

2 tasks

codope added 2 commits February 19, 2025 18:27

Fix partition state update and filter non-existent partitions on restore

333635f

- Mark active bounded splits as finished on partition removal. - Remove pending snapshot splits and unsubscribe dropped log buckets. - Catch PartitionNotExistException during subscribeLog to log a warning.

track removed splits

06d5104

codope force-pushed the handle-partition-remove-event branch from 3814713 to 06d5104 Compare February 19, 2025 13:00

wuchong self-requested a review February 19, 2025 14:25

improve tests

e5a20b2

wuchong reviewed Feb 20, 2025

View reviewed changes

luoyuxia reviewed Feb 21, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[connector] Fix partition state update and filter non-existent partitions on restore #359

[connector] Fix partition state update and filter non-existent partitions on restore #359

codope commented Feb 8, 2025

CLAassistant commented Feb 8, 2025 •

edited

Loading

wuchong Feb 20, 2025

wuchong Feb 20, 2025

wuchong Feb 20, 2025

wuchong commented Feb 20, 2025

luoyuxia left a comment

luoyuxia Feb 21, 2025

luoyuxia Feb 21, 2025

wuchong Feb 21, 2025

luoyuxia Feb 21, 2025

luoyuxia Feb 21, 2025

wuchong commented Feb 21, 2025

		@@ -262,17 +287,42 @@ public Set<TableBucket> removePartitions(Map<Long, String> removedPartitions) {
		// todo, may consider to close the current snapshot reader if

[connector] Fix partition state update and filter non-existent partitions on restore #359

Are you sure you want to change the base?

[connector] Fix partition state update and filter non-existent partitions on restore #359

Conversation

codope commented Feb 8, 2025

Purpose

Changes

Tests

API and Format

Documentation

CLAassistant commented Feb 8, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wuchong commented Feb 20, 2025

luoyuxia left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wuchong commented Feb 21, 2025

CLAassistant commented Feb 8, 2025 •

edited

Loading