FlussConnection getTable method fails due to missing retries when fetching metadata

[swuferhong]

Search before asking

• I searched in the issues and found nothing similar.

Fluss version

0.8.0 (latest release)

Please describe the bug 🐞

FlussConnection getTable method fails due to missing retries when fetching metadata：

java.lang.IllegalArgumentException: table: xxx not found in cluster
	at org.apache.fluss.cluster.Cluster.lambda$getTableOrElseThrow$2(Cluster.java:260)
	at java.util.Optional.orElseThrow(Optional.java:290)
	at org.apache.fluss.cluster.Cluster.getTableOrElseThrow(Cluster.java:257)
	at org.apache.fluss.client.metadata.MetadataUpdater.getTableInfoOrElseThrow(MetadataUpdater.java:100)
	at org.apache.fluss.client.FlussConnection.getTable(FlussConnection.java:103)
	at org.apache.fluss.flink.sink.writer.FlinkSinkWriter.initialize(FlinkSinkWriter.java:112)
	at org.apache.fluss.flink.sink.writer.AppendSinkWriter.initialize(AppendSinkWriter.java:53)
	at org.apache.fluss.flink.sink.FlinkSink.createWriter(FlinkSink.java:63)
	at org.apache.flink.streaming.runtime.operators.sink.StatelessSinkWriterStateHandler.createWriter(StatelessSinkWriterStateHandler.java:39)
	at org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.initializeState(SinkWriterOperator.java:170)
	at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.initializeOperatorState(StreamOperatorStateHandler.java:122)
	at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:285)
	at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:106)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:843)
	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:790)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:747)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:959)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:928)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:751)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:567)
	at java.lang.Thread.run(Thread.java:879)

Solution

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[swuferhong]

@wuchong This bug maybe need to consider when refactor metadata cache in #483

[swuferhong]

Another case is:

java.lang.IllegalStateException: Failed to initialize fluss client connection to server because no bootstrap server is validate. bootstrap servers: [fluss-xxx.com/xxx:80]
	at org.apache.fluss.client.metadata.MetadataUpdater.initializeCluster(MetadataUpdater.java:313)
	at org.apache.fluss.client.metadata.MetadataUpdater.<init>(MetadataUpdater.java:70)
	at org.apache.fluss.client.FlussConnection.<init>(FlussConnection.java:85)
	at org.apache.fluss.client.ConnectionFactory.createConnection(ConnectionFactory.java:66)
	at org.apache.fluss.flink.source.reader.FlinkSourceSplitReader.<init>(FlinkSourceSplitReader.java:127)
	at org.apache.fluss.flink.source.reader.FlinkSourceReader.lambda$new$0(FlinkSourceReader.java:69)
	at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager.createSplitFetcher(SplitFetcherManager.java:196)
	at org.apache.flink.connector.base.source.reader.fetcher.SingleThreadFetcherManager.addSplits(SingleThreadFetcherManager.java:107)
	at org.apache.flink.connector.base.source.reader.SourceReaderBase.addSplits(SourceReaderBase.java:258)
	at org.apache.flink.streaming.api.operators.SourceOperator.open(SourceOperator.java:380)
	at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:107)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:843)
	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:790)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:747)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:959)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:928)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:751)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:567)
	at java.lang.Thread.run(Thread.java:879)
Caused by: java.util.concurrent.ExecutionException: org.apache.fluss.exception.StaleMetadataException: Alive tablet server is empty.
	at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
	at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1928)
	at org.apache.fluss.client.utils.MetadataUtils.sendMetadataRequestAndRebuildCluster(MetadataUtils.java:163)
	at org.apache.fluss.client.utils.MetadataUtils.sendMetadataRequestAndRebuildCluster(MetadataUtils.java:68)
	at org.apache.fluss.client.metadata.MetadataUpdater.tryToInitializeCluster(MetadataUpdater.java:327)
	at org.apache.fluss.client.metadata.MetadataUpdater.initializeCluster(MetadataUpdater.java:296)
	... 19 more
Caused by: org.apache.fluss.exception.StaleMetadataException: Alive tablet server is empty.