| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.flink.configuration; |
| |
| import org.apache.flink.annotation.Internal; |
| import org.apache.flink.annotation.PublicEvolving; |
| import org.apache.flink.annotation.docs.Documentation; |
| import org.apache.flink.configuration.description.Description; |
| import org.apache.flink.util.TimeUtils; |
| |
| import java.time.Duration; |
| |
| import static org.apache.flink.configuration.description.LinkElement.link; |
| import static org.apache.flink.configuration.description.TextElement.code; |
| |
| /** Akka configuration options. */ |
| @PublicEvolving |
| public class AkkaOptions { |
| |
| @Internal |
| @Documentation.ExcludeFromDocumentation("Internal use only") |
| public static final ConfigOption<Boolean> FORCE_RPC_INVOCATION_SERIALIZATION = |
| ConfigOptions.key("pekko.rpc.force-invocation-serialization") |
| .booleanType() |
| .defaultValue(false) |
| .withDeprecatedKeys("akka.rpc.force-invocation-serialization") |
| .withDescription( |
| Description.builder() |
| .text( |
| "Forces the serialization of all RPC invocations (that are not explicitly annotated with %s)." |
| + "This option can be used to find serialization issues in the argument/response types without relying requiring HA setups." |
| + "This option should not be enabled in production.", |
| code("org.apache.flink.runtime.rpc.Local")) |
| .build()); |
| |
| public static boolean isForceRpcInvocationSerializationEnabled(Configuration config) { |
| return config.getOptional(FORCE_RPC_INVOCATION_SERIALIZATION) |
| .orElse( |
| FORCE_RPC_INVOCATION_SERIALIZATION.defaultValue() |
| || System.getProperties() |
| .containsKey(FORCE_RPC_INVOCATION_SERIALIZATION.key())); |
| } |
| |
| /** Flag whether to capture call stacks for RPC ask calls. */ |
| public static final ConfigOption<Boolean> CAPTURE_ASK_CALLSTACK = |
| ConfigOptions.key("pekko.ask.callstack") |
| .booleanType() |
| .defaultValue(true) |
| .withDeprecatedKeys("akka.ask.callstack") |
| .withDescription( |
| "If true, call stack for asynchronous asks are captured. That way, when an ask fails " |
| + "(for example times out), you get a proper exception, describing to the original method call and " |
| + "call site. Note that in case of having millions of concurrent RPC calls, this may add to the " |
| + "memory footprint."); |
| |
| /** Timeout for Pekko ask calls. */ |
| public static final ConfigOption<Duration> ASK_TIMEOUT_DURATION = |
| ConfigOptions.key("pekko.ask.timeout") |
| .durationType() |
| .defaultValue(Duration.ofSeconds(10)) |
| .withDeprecatedKeys("akka.ask.timeout") |
| .withDescription( |
| "Timeout used for all futures and blocking Pekko calls. If Flink fails due to timeouts then you" |
| + " should try to increase this value. Timeouts can be caused by slow machines or a congested network. The" |
| + " timeout value requires a time-unit specifier (ms/s/min/h/d)."); |
| |
| /** @deprecated Use {@link #ASK_TIMEOUT_DURATION} */ |
| @Deprecated |
| public static final ConfigOption<String> ASK_TIMEOUT = |
| ConfigOptions.key(ASK_TIMEOUT_DURATION.key()) |
| .stringType() |
| .defaultValue( |
| TimeUtils.formatWithHighestUnit(ASK_TIMEOUT_DURATION.defaultValue())) |
| .withDescription(ASK_TIMEOUT_DURATION.description()); |
| |
| /** The Pekko tcp connection timeout. */ |
| public static final ConfigOption<String> TCP_TIMEOUT = |
| ConfigOptions.key("pekko.tcp.timeout") |
| .stringType() |
| .defaultValue("20 s") |
| .withDeprecatedKeys("akka.tcp.timeout") |
| .withDescription( |
| "Timeout for all outbound connections. If you should experience problems with connecting to a" |
| + " TaskManager due to a slow network, you should increase this value."); |
| |
| /** Timeout for the startup of the actor system. */ |
| public static final ConfigOption<String> STARTUP_TIMEOUT = |
| ConfigOptions.key("pekko.startup-timeout") |
| .stringType() |
| .noDefaultValue() |
| .withDeprecatedKeys("akka.startup-timeout") |
| .withDescription( |
| "Timeout after which the startup of a remote component is considered being failed."); |
| |
| /** Override SSL support for the Pekko transport. */ |
| public static final ConfigOption<Boolean> SSL_ENABLED = |
| ConfigOptions.key("pekko.ssl.enabled") |
| .booleanType() |
| .defaultValue(true) |
| .withDeprecatedKeys("akka.ssl.enabled") |
| .withDescription( |
| "Turns on SSL for Akka’s remote communication. This is applicable only when the global ssl flag" |
| + " security.ssl.enabled is set to true."); |
| |
| /** Maximum framesize of Pekko messages. */ |
| public static final ConfigOption<String> FRAMESIZE = |
| ConfigOptions.key("pekko.framesize") |
| .stringType() |
| .defaultValue("10485760b") |
| .withDeprecatedKeys("akka.framesize") |
| .withDescription( |
| "Maximum size of messages which are sent between the JobManager and the TaskManagers. If Flink" |
| + " fails because messages exceed this limit, then you should increase it. The message size requires a" |
| + " size-unit specifier."); |
| |
| /** Maximum number of messages until another actor is executed by the same thread. */ |
| public static final ConfigOption<Integer> DISPATCHER_THROUGHPUT = |
| ConfigOptions.key("pekko.throughput") |
| .intType() |
| .defaultValue(15) |
| .withDeprecatedKeys("akka.throughput") |
| .withDescription( |
| "Number of messages that are processed in a batch before returning the thread to the pool. Low" |
| + " values denote a fair scheduling whereas high values can increase the performance at the cost of unfairness."); |
| |
| /** Log lifecycle events. */ |
| public static final ConfigOption<Boolean> LOG_LIFECYCLE_EVENTS = |
| ConfigOptions.key("pekko.log.lifecycle.events") |
| .booleanType() |
| .defaultValue(false) |
| .withDeprecatedKeys("akka.log.lifecycle.events") |
| .withDescription( |
| "Turns on the Pekko’s remote logging of events. Set this value to 'true' in case of debugging."); |
| |
| /** Timeout for all blocking calls that look up remote actors. */ |
| public static final ConfigOption<Duration> LOOKUP_TIMEOUT_DURATION = |
| ConfigOptions.key("pekko.lookup.timeout") |
| .durationType() |
| .defaultValue(Duration.ofSeconds(10)) |
| .withDeprecatedKeys("akka.lookup.timeout") |
| .withDescription( |
| "Timeout used for the lookup of the JobManager. The timeout value has to contain a time-unit" |
| + " specifier (ms/s/min/h/d)."); |
| |
| /** @deprecated use {@link #LOOKUP_TIMEOUT_DURATION} */ |
| @Deprecated |
| public static final ConfigOption<String> LOOKUP_TIMEOUT = |
| ConfigOptions.key(LOOKUP_TIMEOUT_DURATION.key()) |
| .stringType() |
| .defaultValue( |
| TimeUtils.formatWithHighestUnit(LOOKUP_TIMEOUT_DURATION.defaultValue())) |
| .withDescription(LOOKUP_TIMEOUT_DURATION.description()); |
| |
| /** |
| * Timeout for all blocking calls on the client side. |
| * |
| * @deprecated Use the {@code ClientOptions.CLIENT_TIMEOUT} instead. |
| */ |
| @Deprecated |
| public static final ConfigOption<String> CLIENT_TIMEOUT = |
| ConfigOptions.key("akka.client.timeout") |
| .stringType() |
| .defaultValue("60 s") |
| .withDescription( |
| "DEPRECATED: Use the \"client.timeout\" instead." |
| + " Timeout for all blocking calls on the client side."); |
| |
| /** Exit JVM on fatal Pekko errors. */ |
| public static final ConfigOption<Boolean> JVM_EXIT_ON_FATAL_ERROR = |
| ConfigOptions.key("pekko.jvm-exit-on-fatal-error") |
| .booleanType() |
| .defaultValue(true) |
| .withDeprecatedKeys("akka.jvm-exit-on-fatal-error") |
| .withDescription("Exit JVM on fatal Pekko errors."); |
| |
| /** Milliseconds a gate should be closed for after a remote connection was disconnected. */ |
| public static final ConfigOption<Long> RETRY_GATE_CLOSED_FOR = |
| ConfigOptions.key("pekko.retry-gate-closed-for") |
| .longType() |
| .defaultValue(50L) |
| .withDeprecatedKeys("akka.retry-gate-closed-for") |
| .withDescription( |
| "Milliseconds a gate should be closed for after a remote connection was disconnected."); |
| |
| // ================================================== |
| // Configurations for fork-join-executor. |
| // ================================================== |
| |
| public static final ConfigOption<Double> FORK_JOIN_EXECUTOR_PARALLELISM_FACTOR = |
| ConfigOptions.key("pekko.fork-join-executor.parallelism-factor") |
| .doubleType() |
| .defaultValue(2.0) |
| .withDeprecatedKeys("akka.fork-join-executor.parallelism-factor") |
| .withDescription( |
| Description.builder() |
| .text( |
| "The parallelism factor is used to determine thread pool size using the" |
| + " following formula: ceil(available processors * factor). Resulting size" |
| + " is then bounded by the parallelism-min and parallelism-max values.") |
| .build()); |
| |
| public static final ConfigOption<Integer> FORK_JOIN_EXECUTOR_PARALLELISM_MIN = |
| ConfigOptions.key("pekko.fork-join-executor.parallelism-min") |
| .intType() |
| .defaultValue(8) |
| .withDeprecatedKeys("akka.fork-join-executor.parallelism-min") |
| .withDescription( |
| Description.builder() |
| .text( |
| "Min number of threads to cap factor-based parallelism number to.") |
| .build()); |
| |
| public static final ConfigOption<Integer> FORK_JOIN_EXECUTOR_PARALLELISM_MAX = |
| ConfigOptions.key("pekko.fork-join-executor.parallelism-max") |
| .intType() |
| .defaultValue(64) |
| .withDeprecatedKeys("akka.fork-join-executor.parallelism-max") |
| .withDescription( |
| Description.builder() |
| .text( |
| "Max number of threads to cap factor-based parallelism number to.") |
| .build()); |
| |
| // ================================================== |
| // Configurations for client-socket-work-pool. |
| // ================================================== |
| |
| public static final ConfigOption<Integer> CLIENT_SOCKET_WORKER_POOL_SIZE_MIN = |
| ConfigOptions.key("pekko.client-socket-worker-pool.pool-size-min") |
| .intType() |
| .defaultValue(1) |
| .withDeprecatedKeys("akka.client-socket-worker-pool.pool-size-min") |
| .withDescription( |
| Description.builder() |
| .text("Min number of threads to cap factor-based number to.") |
| .build()); |
| |
| public static final ConfigOption<Integer> CLIENT_SOCKET_WORKER_POOL_SIZE_MAX = |
| ConfigOptions.key("pekko.client-socket-worker-pool.pool-size-max") |
| .intType() |
| .defaultValue(2) |
| .withDeprecatedKeys("akka.client-socket-worker-pool.pool-size-max") |
| .withDescription( |
| Description.builder() |
| .text("Max number of threads to cap factor-based number to.") |
| .build()); |
| |
| public static final ConfigOption<Double> CLIENT_SOCKET_WORKER_POOL_SIZE_FACTOR = |
| ConfigOptions.key("pekko.client-socket-worker-pool.pool-size-factor") |
| .doubleType() |
| .defaultValue(1.0) |
| .withDeprecatedKeys("akka.client-socket-worker-pool.pool-size-factor") |
| .withDescription( |
| Description.builder() |
| .text( |
| "The pool size factor is used to determine thread pool size" |
| + " using the following formula: ceil(available processors * factor)." |
| + " Resulting size is then bounded by the pool-size-min and" |
| + " pool-size-max values.") |
| .build()); |
| |
| // ================================================== |
| // Configurations for server-socket-work-pool. |
| // ================================================== |
| |
| public static final ConfigOption<Integer> SERVER_SOCKET_WORKER_POOL_SIZE_MIN = |
| ConfigOptions.key("pekko.server-socket-worker-pool.pool-size-min") |
| .intType() |
| .defaultValue(1) |
| .withDeprecatedKeys("akka.server-socket-worker-pool.pool-size-min") |
| .withDescription( |
| Description.builder() |
| .text("Min number of threads to cap factor-based number to.") |
| .build()); |
| |
| public static final ConfigOption<Integer> SERVER_SOCKET_WORKER_POOL_SIZE_MAX = |
| ConfigOptions.key("pekko.server-socket-worker-pool.pool-size-max") |
| .intType() |
| .defaultValue(2) |
| .withDeprecatedKeys("akka.server-socket-worker-pool.pool-size-max") |
| .withDescription( |
| Description.builder() |
| .text("Max number of threads to cap factor-based number to.") |
| .build()); |
| |
| public static final ConfigOption<Double> SERVER_SOCKET_WORKER_POOL_SIZE_FACTOR = |
| ConfigOptions.key("pekko.server-socket-worker-pool.pool-size-factor") |
| .doubleType() |
| .defaultValue(1.0) |
| .withDeprecatedKeys("akka.server-socket-worker-pool.pool-size-factor") |
| .withDescription( |
| Description.builder() |
| .text( |
| "The pool size factor is used to determine thread pool size" |
| + " using the following formula: ceil(available processors * factor)." |
| + " Resulting size is then bounded by the pool-size-min and" |
| + " pool-size-max values.") |
| .build()); |
| |
| // ================================================== |
| // Deprecated options |
| // ================================================== |
| |
| /** |
| * The Akka death watch heartbeat interval. |
| * |
| * @deprecated Don't use this option anymore. It has no effect on Flink. |
| */ |
| @Deprecated |
| public static final ConfigOption<String> WATCH_HEARTBEAT_INTERVAL = |
| ConfigOptions.key("akka.watch.heartbeat.interval") |
| .stringType() |
| .defaultValue(ASK_TIMEOUT.defaultValue()) |
| .withDescription( |
| Description.builder() |
| .text( |
| "Heartbeat interval for Akka’s DeathWatch mechanism to detect dead TaskManagers. If" |
| + " TaskManagers are wrongly marked dead because of lost or delayed heartbeat messages, then you" |
| + " should decrease this value or increase akka.watch.heartbeat.pause. A thorough description of" |
| + " Akka’s DeathWatch can be found %s", |
| link( |
| "https://pekko.apache.org/docs/pekko/current/remoting-artery.html#failure-detector", |
| "here")) |
| .build()); |
| |
| /** |
| * The maximum acceptable Akka death watch heartbeat pause. |
| * |
| * @deprecated Don't use this option anymore. It has no effect on Flink. |
| */ |
| @Deprecated |
| public static final ConfigOption<String> WATCH_HEARTBEAT_PAUSE = |
| ConfigOptions.key("akka.watch.heartbeat.pause") |
| .stringType() |
| .defaultValue("60 s") |
| .withDescription( |
| Description.builder() |
| .text( |
| "Acceptable heartbeat pause for Akka’s DeathWatch mechanism. A low value does not allow an" |
| + " irregular heartbeat. If TaskManagers are wrongly marked dead because of lost or delayed" |
| + " heartbeat messages, then you should increase this value or decrease akka.watch.heartbeat.interval." |
| + " Higher value increases the time to detect a dead TaskManager. A thorough description of Akka’s" |
| + " DeathWatch can be found %s", |
| link( |
| "https://pekko.apache.org/docs/pekko/current/remoting-artery.html#failure-detector", |
| "here")) |
| .build()); |
| |
| /** |
| * Detection threshold for the phi accrual watch failure detector. |
| * |
| * @deprecated Don't use this option anymore. It has no effect on Flink. |
| */ |
| @Deprecated |
| public static final ConfigOption<Integer> WATCH_THRESHOLD = |
| ConfigOptions.key("akka.watch.threshold") |
| .intType() |
| .defaultValue(12) |
| .withDescription( |
| Description.builder() |
| .text( |
| "Threshold for the DeathWatch failure detector. A low value is prone to false positives whereas" |
| + " a high value increases the time to detect a dead TaskManager. A thorough description of Akka’s" |
| + " DeathWatch can be found %s", |
| link( |
| "https://pekko.apache.org/docs/pekko/current/remoting-artery.html#failure-detector", |
| "here")) |
| .build()); |
| } |
