Repository navigation
[Bug] New Partiions created will cause Flink job failover when partition discovery is enabled while scanning #288
Description
Search before asking
- I searched in the issues and found nothing similar.
Fluss version
0.5.0
Minimal reproduce step
Create a partitioned table and a Flink job to subscribe it.. After new partitions is created, it may well throw
Caused by: com.alibaba.fluss.exception.FlussRuntimeException: Leader not found for table bucket: TableBucket{tableId=0, partitionId=60, bucket=16}
at com.alibaba.fluss.client.metadata.MetadataUpdater.leaderFor(MetadataUpdater.java:122)
at com.alibaba.fluss.client.admin.FlussAdmin.prepareListOffsetsRequests(FlussAdmin.java:295)
at com.alibaba.fluss.client.admin.FlussAdmin.listOffsets(FlussAdmin.java:253)
at com.alibaba.fluss.connector.flink.source.enumerator.initializer.BucketOffsetsRetrieverImpl.listOffsets(BucketOffsetsRetrieverImpl.java:72)
at com.alibaba.fluss.connector.flink.source.enumerator.initializer.BucketOffsetsRetrieverImpl.latestOffsets(BucketOffsetsRetrieverImpl.java:49)
at com.alibaba.fluss.connector.flink.source.enumerator.initializer.LatestOffsetsInitializer.getBucketOffsets(LatestOffsetsInitializer.java:38)
at com.alibaba.fluss.connector.flink.source.enumerator.FlinkSourceEnumerator.getLogSplit(FlinkSourceEnumerator.java:449)
at com.alibaba.fluss.connector.flink.source.enumerator.FlinkSourceEnumerator.initLogTablePartitionSplits(FlinkSourceEnumerator.java:356)
at com.alibaba.fluss.connector.flink.source.enumerator.FlinkSourceEnumerator.initPartitionedSplits(FlinkSourceEnumerator.java:348)
at com.alibaba.fluss.connector.flink.source.enumerator.FlinkSourceEnumerator.lambda$checkPartitionChanges$2(FlinkSourceEnumerator.java:296)
at org.apache.flink.runtime.source.coordinator.ExecutorNotifier.lambda$notifyReadyAsync$2(ExecutorNotifier.java:80)
It'll cause job failover
What doesn't meet your expectations?
It's in the expectation that throws an exception since the leader may well still not be elected although partition has been created...Although it'll running normally after failover, we'd better to handle this cause to avoid failover again and again in every partition created.
Anything else?
No response
Are you willing to submit a PR?
- I'm willing to submit a PR!
Activity
luoyuxia commentedon Jan 9, 2025
Seems main branch has introduced a try machesim with max try time 5... It may be still too short to wait leader ready if response of update metada return quickly..
Let me check whether the problem can still happen in latest branch..
Btw, I think we should follow kafak's desgin which will retry foerver if
LeaderNotAvailableExceptionis thrown..luoyuxia commentedon Jan 10, 2025
It turns out the exception can still happen:
luoyuxia commentedon Jan 15, 2025
Btw, I think we should backoff when try again..
luoyuxia commentedon Jan 15, 2025
cc @swuferhong @loserwang1024
loserwang1024 commentedon Jan 15, 2025
I wonder it will also happens if a leader is lost and the new leader is not be decided? It seems kafka add triable to ListOffsets in https://issues.apache.org/jira/browse/KAFKA-14821 in kafka 3.5.0.
luoyuxia commentedon Feb 20, 2025
@loserwang1024 What's the progress of this issue as v0.6 is approaching
feat: support kv tables in CPP (apache#288)
feat: support kv tables in CPP (#288)