[hotfix][client] Retry to initial cluster when encounter TimeoutException

[swuferhong]

Purpose

Linked issue: close #xxx

java.lang.IllegalStateException: Failed to initialize fluss client connection to bootstrap servers: [xxx..com/xxx:80]. 
Reason: null
	at org.apache.fluss.client.metadata.MetadataUpdater.initializeCluster(MetadataUpdater.java:319)
	at org.apache.fluss.client.metadata.MetadataUpdater.<init>(MetadataUpdater.java:72)
	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:842)
	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:789)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:746)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:968)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:937)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:754)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:569)
	at java.lang.Thread.run(Thread.java:879)
Caused by: java.util.concurrent.TimeoutException
	at java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1784)
	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:372)
	at org.apache.fluss.client.metadata.MetadataUpdater.tryToInitializeClusterWithRetries(MetadataUpdater.java:335)
	at org.apache.fluss.client.metadata.MetadataUpdater.initializeCluster(MetadataUpdater.java:292)
	... 19 more

Brief change log

Tests

API and Format

Documentation