[Bug] New Partiions created will cause Flink job failover when partition discovery is enabled while scanning

[luoyuxia]

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!

[luoyuxia]

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 LeaderNotAvailableException is thrown..

[luoyuxia]

It turns out the exception can still happen:

Caused by: com.alibaba.fluss.exception.FlussRuntimeException: Leader not found after retry  5 times for table bucket: TableBucket{tableId=0, partitionId=2, bucket=23}
	at com.alibaba.fluss.client.metadata.MetadataUpdater.leaderFor(MetadataUpdater.java:135)
	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)
	... 7 more

[luoyuxia]

Btw, I think we should backoff when try again..

[luoyuxia]

cc @swuferhong @loserwang1024

[loserwang1024]

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.