聊聊flink的RestClusterClientConfiguration

序本文主要研究一下flink的RestClusterClientConfigurationRestClusterClientConfigurationflink-release-1.7.2/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClientConfiguration.javapublic final class RestClusterClientConfiguration { private final RestClientConfiguration restClientConfiguration; private final long awaitLeaderTimeout; private final int retryMaxAttempts; private final long retryDelay; private RestClusterClientConfiguration( final RestClientConfiguration endpointConfiguration, final long awaitLeaderTimeout, final int retryMaxAttempts, final long retryDelay) { checkArgument(awaitLeaderTimeout >= 0, “awaitLeaderTimeout must be equal to or greater than 0”); checkArgument(retryMaxAttempts >= 0, “retryMaxAttempts must be equal to or greater than 0”); checkArgument(retryDelay >= 0, “retryDelay must be equal to or greater than 0”); this.restClientConfiguration = Preconditions.checkNotNull(endpointConfiguration); this.awaitLeaderTimeout = awaitLeaderTimeout; this.retryMaxAttempts = retryMaxAttempts; this.retryDelay = retryDelay; } public RestClientConfiguration getRestClientConfiguration() { return restClientConfiguration; } /** * @see RestOptions#AWAIT_LEADER_TIMEOUT / public long getAwaitLeaderTimeout() { return awaitLeaderTimeout; } /* * @see RestOptions#RETRY_MAX_ATTEMPTS / public int getRetryMaxAttempts() { return retryMaxAttempts; } /* * @see RestOptions#RETRY_DELAY / public long getRetryDelay() { return retryDelay; } public static RestClusterClientConfiguration fromConfiguration(Configuration config) throws ConfigurationException { RestClientConfiguration restClientConfiguration = RestClientConfiguration.fromConfiguration(config); final long awaitLeaderTimeout = config.getLong(RestOptions.AWAIT_LEADER_TIMEOUT); final int retryMaxAttempts = config.getInteger(RestOptions.RETRY_MAX_ATTEMPTS); final long retryDelay = config.getLong(RestOptions.RETRY_DELAY); return new RestClusterClientConfiguration(restClientConfiguration, awaitLeaderTimeout, retryMaxAttempts, retryDelay); }}RestClusterClientConfiguration除了RestClientConfiguration外，还有3个属性，分别是awaitLeaderTimeout、retryMaxAttempts、retryDelay；awaitLeaderTimeout读取的是rest.await-leader-timeout配置，默认是30秒；retryMaxAttempts读取的是rest.retry.max-attempts配置，默认是20；retryDelay读取的是rest.retry.delay配置，默认是3秒RestClusterClientflink-release-1.7.2/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.javapublic class RestClusterClient<T> extends ClusterClient<T> implements NewClusterClient { private final RestClusterClientConfiguration restClusterClientConfiguration; private final RestClient restClient; private final ExecutorService executorService = Executors.newFixedThreadPool(4, new ExecutorThreadFactory(“Flink-RestClusterClient-IO”)); private final WaitStrategy waitStrategy; private final T clusterId; private final LeaderRetrievalService webMonitorRetrievalService; private final LeaderRetrievalService dispatcherRetrievalService; private final LeaderRetriever webMonitorLeaderRetriever = new LeaderRetriever(); private final LeaderRetriever dispatcherLeaderRetriever = new LeaderRetriever(); /* ExecutorService to run operations that can be retried on exceptions. */ private ScheduledExecutorService retryExecutorService; //…… private <C> CompletableFuture<C> retry( CheckedSupplier<CompletableFuture<C>> operation, Predicate<Throwable> retryPredicate) { return FutureUtils.retryWithDelay( CheckedSupplier.unchecked(operation), restClusterClientConfiguration.getRetryMaxAttempts(), Time.milliseconds(restClusterClientConfiguration.getRetryDelay()), retryPredicate, new ScheduledExecutorServiceAdapter(retryExecutorService)); } @VisibleForTesting CompletableFuture<URL> getWebMonitorBaseUrl() { return FutureUtils.orTimeout( webMonitorLeaderRetriever.getLeaderFuture(), restClusterClientConfiguration.getAwaitLeaderTimeout(), TimeUnit.MILLISECONDS) .thenApplyAsync(leaderAddressSessionId -> { final String url = leaderAddressSessionId.f0; try { return new URL(url); } catch (MalformedURLException e) { throw new IllegalArgumentException(“Could not parse URL from " + url, e); } }, executorService); } //……}RestClusterClient的构造器会从使用RestClusterClientConfiguration.fromConfiguration(configuration)方法从Configuration构建RestClusterClientConfigurationretry方法内部使用的是FutureUtils.retryWithDelay方法，其retries参数使用的是restClusterClientConfiguration.getRetryMaxAttempts()，retryDelay参数使用的是Time.milliseconds(restClusterClientConfiguration.getRetryDelay())getWebMonitorBaseUrl方法内部使用的是FutureUtils.orTimeout方法，其timeout参数使用的是restClusterClientConfiguration.getAwaitLeaderTimeout()小结RestClusterClientConfiguration除了RestClientConfiguration外，还有3个属性，分别是awaitLeaderTimeout、retryMaxAttempts、retryDelayawaitLeaderTimeout读取的是rest.await-leader-timeout配置，默认是30秒；retryMaxAttempts读取的是rest.retry.max-attempts配置，默认是20；retryDelay读取的是rest.retry.delay配置，默认是3秒RestClusterClient的etry方法内部使用的是FutureUtils.retryWithDelay方法，其retries参数使用的是restClusterClientConfiguration.getRetryMaxAttempts()，retryDelay参数使用的是Time.milliseconds(restClusterClientConfiguration.getRetryDelay())；getWebMonitorBaseUrl方法内部使用的是FutureUtils.orTimeout方法，其timeout参数使用的是restClusterClientConfiguration.getAwaitLeaderTimeout()docRestClusterClientConfiguration