fxbing commented on code in PR #4608:
URL: https://github.com/apache/flink-cdc/pull/4608#discussion_r4537971136
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSourceReader.java:
##########
@@ -243,7 +263,7 @@ protected void onSplitFinished(Map<String, FlussSplitState>
finishedSplitIds) {
private void activateSplits(List<FlussSplitBase> splits) {
List<FlussSplitBase> activeSplits = new ArrayList<>();
for (FlussSplitBase split : splits) {
- if (isActive(split.getTablePath())) {
+ if (isSplitActive(split)) {
Review Comment:
@litiliu Thanks, this handles the case where the partition was dropped. My
concern is what happens just after a new partition is created. Suppose the
enumerator discovers today’s `orders` partition through TabletServer A and
assigns bucket 0. The reader’s `subscribe()` and subsequent
`listPartitionInfos()` can both reach TabletServer B. If B hasn’t cached the
new partition yet and its ZooKeeper replica is still catching up, the ID lookup
can throw `PartitionNotExistException`, and the list call can succeed without
including the partition.
The reader would then finish the split. Even after B catches up, the
enumerator still considers the partition assigned, so it won’t reassign bucket
0. I injected those two responses and verified that the split was still missing
after checkpoint restore. Does Fluss guarantee that once A has returned a newly
created partition, B must also see it on these reads?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
