Package org.apache.flink.api.common

Class ExecutionConfig

  • All Implemented Interfaces:
    Serializable, Archiveable<ArchivedExecutionConfig>
    @Public
public class ExecutionConfig
extends Object
implements Serializable, Archiveable<ArchivedExecutionConfig>
    A config to define the behavior of the program execution. It allows to define (among other options) the following settings:
    • The default parallelism of the program, i.e., how many parallel tasks to use for all functions that do not define a specific value directly.
    • The number of retries in the case of failed executions.
    • The delay between execution retries.
    • The ExecutionMode of the program: Batch or Pipelined. The default execution mode is ExecutionMode.PIPELINED
    • Enabling or disabling the "closure cleaner". The closure cleaner pre-processes the implementations of functions. In case they are (anonymous) inner classes, it removes unused references to the enclosing class to fix certain serialization-related problems and to reduce the size of the closure.
    • The config allows to register types and serializers to increase the efficiency of handling generic types and POJOs. This is usually only needed when the functions return not only the types declared in their signature, but also subclasses of those types.
    See Also:
    Serialized Form

    • Field Detail

      • PARALLELISM_AUTO_MAX

        @Deprecated
public static final int PARALLELISM_AUTO_MAX
        Deprecated.
        The constant to use for the parallelism, if the system should use the number of currently available slots.
        See Also:
        Constant Field Values

      • PARALLELISM_DEFAULT

        public static final int PARALLELISM_DEFAULT
        The flag value indicating use of the default parallelism. This value can be used to reset the parallelism back to the default state.
        See Also:
        Constant Field Values

      • PARALLELISM_UNKNOWN

        public static final int PARALLELISM_UNKNOWN
        The flag value indicating an unknown or unset parallelism. This value is not a valid parallelism and indicates that the parallelism should remain unchanged.
        See Also:
        Constant Field Values

    • Constructor Detail

      • ExecutionConfig

        public ExecutionConfig()

      • ExecutionConfig

        @Internal
public ExecutionConfig​(Configuration configuration)

    • Method Detail

      • enableClosureCleaner

        public ExecutionConfig enableClosureCleaner()
        Enables the ClosureCleaner. This analyzes user code functions and sets fields to null that are not used. This will in most cases make closures or anonymous inner classes serializable that where not serializable due to some Scala or Java implementation artifact. User code must be serializable because it needs to be sent to worker nodes.

      • isClosureCleanerEnabled

        public boolean isClosureCleanerEnabled()
        Returns whether the ClosureCleaner is enabled.
        See Also:
        enableClosureCleaner()

      • setAutoWatermarkInterval

        @PublicEvolving
public ExecutionConfig setAutoWatermarkInterval​(long interval)
        Sets the interval of the automatic watermark emission. Watermarks are used throughout the streaming system to keep track of the progress of time. They are used, for example, for time based windowing.

        Setting an interval of 0 will disable periodic watermark emission.

        Parameters:
        interval - The interval between watermarks in milliseconds.

      • getAutoWatermarkInterval

        @PublicEvolving
public long getAutoWatermarkInterval()
        Returns the interval of the automatic watermark emission.
        See Also:
        setAutoWatermarkInterval(long)

      • setLatencyTrackingInterval

        @PublicEvolving
public ExecutionConfig setLatencyTrackingInterval​(long interval)
        Interval for sending latency tracking marks from the sources to the sinks. Flink will send latency tracking marks from the sources at the specified interval.

        Setting a tracking interval <= 0 disables the latency tracking.

        Parameters:
        interval - Interval in milliseconds.

      • getLatencyTrackingInterval

        @PublicEvolving
public long getLatencyTrackingInterval()
        Returns the latency tracking interval.
        Returns:
        The latency tracking interval in milliseconds

      • isLatencyTrackingConfigured

        @Internal
public boolean isLatencyTrackingConfigured()

      • isPeriodicMaterializeEnabled

        @Internal
public boolean isPeriodicMaterializeEnabled()

      • enablePeriodicMaterialize

        @Internal
public void enablePeriodicMaterialize​(boolean enabled)

      • getPeriodicMaterializeIntervalMillis

        @Internal
public long getPeriodicMaterializeIntervalMillis()

      • setPeriodicMaterializeIntervalMillis

        @Internal
public void setPeriodicMaterializeIntervalMillis​(java.time.Duration periodicMaterializeInterval)

      • getMaterializationMaxAllowedFailures

        @Internal
public int getMaterializationMaxAllowedFailures()

      • setMaterializationMaxAllowedFailures

        @Internal
public void setMaterializationMaxAllowedFailures​(int materializationMaxAllowedFailures)

      • getParallelism

        public int getParallelism()
        Gets the parallelism with which operation are executed by default. Operations can individually override this value to use a specific parallelism.

        Other operations may need to run with a different parallelism - for example calling a reduce operation over the entire data set will involve an operation that runs with a parallelism of one (the final reduce to the single result value).

        Returns:
        The parallelism used by operations, unless they override that value. This method returns PARALLELISM_DEFAULT if the environment's default parallelism should be used.

      • setParallelism

        public ExecutionConfig setParallelism​(int parallelism)
        Sets the parallelism for operations executed through this environment. Setting a parallelism of x here will cause all operators (such as join, map, reduce) to run with x parallel instances.

        This method overrides the default parallelism for this environment. The local execution environment uses by default a value equal to the number of hardware contexts (CPU cores / threads). When executing the program via the command line client from a JAR file, the default parallelism is the one configured for that setup.

        Parameters:
        parallelism - The parallelism to use

      • resetParallelism

        @Internal
public void resetParallelism()

      • getMaxParallelism

        @PublicEvolving
public int getMaxParallelism()
        Gets the maximum degree of parallelism defined for the program.

        The maximum degree of parallelism specifies the upper limit for dynamic scaling. It also defines the number of key groups used for partitioned state.

        Returns:
        Maximum degree of parallelism

      • setMaxParallelism

        @PublicEvolving
public void setMaxParallelism​(int maxParallelism)
        Sets the maximum degree of parallelism defined for the program.

        The maximum degree of parallelism specifies the upper limit for dynamic scaling. It also defines the number of key groups used for partitioned state.

        Parameters:
        maxParallelism - Maximum degree of parallelism to be used for the program.

      • getTaskCancellationInterval

        public long getTaskCancellationInterval()
        Gets the interval (in milliseconds) between consecutive attempts to cancel a running task.

      • setTaskCancellationInterval

        public ExecutionConfig setTaskCancellationInterval​(long interval)
        Sets the configuration parameter specifying the interval (in milliseconds) between consecutive attempts to cancel a running task.
        Parameters:
        interval - the interval (in milliseconds).

      • getTaskCancellationTimeout

        @PublicEvolving
public long getTaskCancellationTimeout()
        Returns the timeout (in milliseconds) after which an ongoing task cancellation leads to a fatal TaskManager error.

        The value 0 means that the timeout is disabled. In this case a stuck cancellation will not lead to a fatal error.

      • setTaskCancellationTimeout

        @PublicEvolving
public ExecutionConfig setTaskCancellationTimeout​(long timeout)
        Sets the timeout (in milliseconds) after which an ongoing task cancellation is considered failed, leading to a fatal TaskManager error.

        The cluster default is configured via TaskManagerOptions.TASK_CANCELLATION_TIMEOUT.

        The value 0 disables the timeout. In this case a stuck cancellation will not lead to a fatal error.

        Parameters:
        timeout - The task cancellation timeout (in milliseconds).

      • setRestartStrategy

        @Deprecated
@PublicEvolving
public void setRestartStrategy​(RestartStrategies.RestartStrategyConfiguration restartStrategyConfiguration)
        Deprecated.
        The method is marked as deprecated because starting from Flink 1.19, the usage of all complex Java objects related to configuration, including their getter and setter methods, should be replaced by ConfigOption. In a future major version of Flink, this method will be removed entirely. It is recommended to switch to using the ConfigOptions provided by RestartStrategyOptions for configuring restart strategies.
        Sets the restart strategy to be used for recovery. 
        
 ExecutionConfig config = env.getConfig();

 config.setRestartStrategy(RestartStrategies.fixedDelayRestart(
 	10,  // number of retries
 	1000 // delay between retries));
        Parameters:
        restartStrategyConfiguration - Configuration defining the restart strategy to use

      • getRestartStrategy

        @Deprecated
@PublicEvolving
public RestartStrategies.RestartStrategyConfiguration getRestartStrategy()
        Deprecated.
        The method is marked as deprecated because starting from Flink 1.19, the usage of all complex Java objects related to configuration, including their getter and setter methods, should be replaced by ConfigOption. In a future major version of Flink, this method will be removed entirely. It is recommended to switch to using the ConfigOptions provided by RestartStrategyOptions for configuring restart strategies.
        Returns the restart strategy which has been set for the current job.
        Returns:
        The specified restart configuration

      • getNumberOfExecutionRetries

        @Deprecated
public int getNumberOfExecutionRetries()
        Deprecated.
        Should no longer be used because it is subsumed by RestartStrategyConfiguration
        Gets the number of times the system will try to re-execute failed tasks. A value of -1 indicates that the system default value (as defined in the configuration) should be used.
        Returns:
        The number of times the system will try to re-execute failed tasks.

      • getExecutionRetryDelay

        @Deprecated
public long getExecutionRetryDelay()
        Deprecated.
        Should no longer be used because it is subsumed by RestartStrategyConfiguration
        Returns the delay between execution retries.
        Returns:
        The delay between successive execution retries in milliseconds.

      • setExecutionMode

        @Deprecated
public void setExecutionMode​(ExecutionMode executionMode)
        Deprecated.
        The ExecutionMode is deprecated because it's only used in DataSet APIs. All Flink DataSet APIs are deprecated since Flink 1.18 and will be removed in a future Flink major version. You can still build your application in DataSet, but you should move to either the DataStream and/or Table API.
        Sets the execution mode to execute the program. The execution mode defines whether data exchanges are performed in a batch or on a pipelined manner.

        The default execution mode is ExecutionMode.PIPELINED.

        Parameters:
        executionMode - The execution mode to use.
        See Also:
        FLIP-131: Consolidate the user-facing Dataflow SDKs/APIs (and deprecate the DataSet API

      • setDefaultInputDependencyConstraint

        @PublicEvolving
@Deprecated
public void setDefaultInputDependencyConstraint​(InputDependencyConstraint ignored)
        Deprecated.
        due to the deprecation of InputDependencyConstraint.
        This method is deprecated. It was used to set the InputDependencyConstraint utilized by the old scheduler implementations which got removed as part of FLINK-20589. The current implementation has no effect.
        Parameters:
        ignored - Ignored parameter.

      • getDefaultInputDependencyConstraint

        @PublicEvolving
@Deprecated
public InputDependencyConstraint getDefaultInputDependencyConstraint()
        Deprecated.
        due to the deprecation of InputDependencyConstraint.
        This method is deprecated. It was used to return the InputDependencyConstraint utilized by the old scheduler implementations. These implementations were removed as part of FLINK-20589.
        Returns:
        The previous default constraint InputDependencyConstraint.ANY.

      • enableForceKryo

        @Deprecated
public void enableForceKryo()
        Deprecated.
        Configure serialization behavior through hard codes is deprecated, because you need to modify the codes when upgrading job version. You should configure this by config option PipelineOptions.FORCE_KRYO.
        Force TypeExtractor to use Kryo serializer for POJOS even though we could analyze as POJO. In some cases this might be preferable. For example, when using interfaces with subclasses that cannot be analyzed as POJO.
        See Also:
        FLIP-398: Improve Serialization Configuration And Usage In Flink

      • disableGenericTypes

        @Deprecated
public void disableGenericTypes()
        Deprecated.
        Configure serialization behavior through hard codes is deprecated, because you need to modify the codes when upgrading job version. You should configure this by config option PipelineOptions.GENERIC_TYPES.
        Disables the use of generic types (types that would be serialized via Kryo). If this option is used, Flink will throw an UnsupportedOperationException whenever it encounters a data type that would go through Kryo for serialization.

        Disabling generic types can be helpful to eagerly find and eliminate the use of types that would go through Kryo serialization during runtime. Rather than checking types individually, using this option will throw exceptions eagerly in the places where generic types are used.

        Important: We recommend to use this option only during development and pre-production phases, not during actual production use. The application program and/or the input data may be such that new, previously unseen, types occur at some point. In that case, setting this option would cause the program to fail.

        See Also:
        FLIP-398: Improve Serialization Configuration And Usage In Flink, enableGenericTypes()

      • enableAutoGeneratedUIDs

        public void enableAutoGeneratedUIDs()
        Enables the Flink runtime to auto-generate UID's for operators.
        See Also:
        disableAutoGeneratedUIDs()

      • disableAutoGeneratedUIDs

        public void disableAutoGeneratedUIDs()
        Disables auto-generated UIDs. Forces users to manually specify UIDs on DataStream applications.

        It is highly recommended that users specify UIDs before deploying to production since they are used to match state in savepoints to operators in a job. Because auto-generated ID's are likely to change when modifying a job, specifying custom IDs allow an application to evolve overtime without discarding state.

      • enableObjectReuse

        public ExecutionConfig enableObjectReuse()
        Enables reusing objects that Flink internally uses for deserialization and passing data to user-code functions. Keep in mind that this can lead to bugs when the user-code function of an operation is not aware of this behaviour.

      • disableObjectReuse

        public ExecutionConfig disableObjectReuse()
        Disables reusing objects that Flink internally uses for deserialization and passing data to user-code functions. @see #enableObjectReuse()

      • isObjectReuseEnabled

        public boolean isObjectReuseEnabled()
        Returns whether object reuse has been enabled or disabled. @see #enableObjectReuse()

      • setGlobalJobParameters

        public void setGlobalJobParameters​(ExecutionConfig.GlobalJobParameters globalJobParameters)
        Register a custom, serializable user configuration object.
        Parameters:
        globalJobParameters - Custom user configuration object

      • addDefaultKryoSerializer

        @Deprecated
public <T extends com.esotericsoftware.kryo.Serializer<?> & Serializable> void addDefaultKryoSerializer​(Class<?> type,
                                                                                                        T serializer)
        Deprecated.
        Register data types and serializers through hard codes is deprecated, because you need to modify the codes when upgrading job version. You should configure this by config option PipelineOptions.SERIALIZATION_CONFIG.
        Adds a new Kryo default serializer to the Runtime.

        Note that the serializer instance must be serializable (as defined by java.io.Serializable), because it may be distributed to the worker nodes by java serialization.

        Parameters:
        type - The class of the types serialized with the given serializer.
        serializer - The serializer to use.
        See Also:
        FLIP-398: Improve Serialization Configuration And Usage In Flink

      • addDefaultKryoSerializer

        @Deprecated
public void addDefaultKryoSerializer​(Class<?> type,
                                     Class<? extends com.esotericsoftware.kryo.Serializer<?>> serializerClass)
        Deprecated.
        Register data types and serializers through hard codes is deprecated, because you need to modify the codes when upgrading job version. You should configure this by config option PipelineOptions.SERIALIZATION_CONFIG.
        Adds a new Kryo default serializer to the Runtime.
        Parameters:
        type - The class of the types serialized with the given serializer.
        serializerClass - The class of the serializer to use.
        See Also:
        FLIP-398: Improve Serialization Configuration And Usage In Flink

      • registerTypeWithKryoSerializer

        @Deprecated
public <T extends com.esotericsoftware.kryo.Serializer<?> & Serializable> void registerTypeWithKryoSerializer​(Class<?> type,
                                                                                                              T serializer)
        Deprecated.
        Register data types and serializers through hard codes is deprecated, because you need to modify the codes when upgrading job version. You should configure this by config option PipelineOptions.SERIALIZATION_CONFIG.
        Registers the given type with a Kryo Serializer.

        Note that the serializer instance must be serializable (as defined by java.io.Serializable), because it may be distributed to the worker nodes by java serialization.

        Parameters:
        type - The class of the types serialized with the given serializer.
        serializer - The serializer to use.
        See Also:
        FLIP-398: Improve Serialization Configuration And Usage In Flink

      • registerTypeWithKryoSerializer

        @Deprecated
public void registerTypeWithKryoSerializer​(Class<?> type,
                                           Class<? extends com.esotericsoftware.kryo.Serializer> serializerClass)
        Deprecated.
        Register data types and serializers through hard codes is deprecated, because you need to modify the codes when upgrading job version. You should configure this by config option PipelineOptions.SERIALIZATION_CONFIG.
        Registers the given Serializer via its class as a serializer for the given type at the KryoSerializer.
        Parameters:
        type - The class of the types serialized with the given serializer.
        serializerClass - The class of the serializer to use.
        See Also:
        FLIP-398: Improve Serialization Configuration And Usage In Flink

      • registerPojoType

        @Deprecated
public void registerPojoType​(Class<?> type)
        Deprecated.
        Register data types and serializers through hard codes is deprecated, because you need to modify the codes when upgrading job version. You should configure this by config option PipelineOptions.SERIALIZATION_CONFIG.
        Registers the given type with the serialization stack. If the type is eventually serialized as a POJO, then the type is registered with the POJO serializer. If the type ends up being serialized with Kryo, then it will be registered at Kryo to make sure that only tags are written.
        Parameters:
        type - The class of the type to register.
        See Also:
        FLIP-398: Improve Serialization Configuration And Usage In Flink

      • registerKryoType

        @Deprecated
public void registerKryoType​(Class<?> type)
        Deprecated.
        Register data types and serializers through hard codes is deprecated, because you need to modify the codes when upgrading job version. You should configure this by config option PipelineOptions.SERIALIZATION_CONFIG.
        Registers the given type with the serialization stack. If the type is eventually serialized as a POJO, then the type is registered with the POJO serializer. If the type ends up being serialized with Kryo, then it will be registered at Kryo to make sure that only tags are written.
        Parameters:
        type - The class of the type to register.
        See Also:
        FLIP-398: Improve Serialization Configuration And Usage In Flink

      • isAutoTypeRegistrationDisabled

        @Deprecated
public boolean isAutoTypeRegistrationDisabled()
        Deprecated.
        The method is deprecated because it's only used in DataSet API. All Flink DataSet APIs are deprecated since Flink 1.18 and will be removed in a future Flink major version. You can still build your application in DataSet, but you should move to either the DataStream and/or Table API.
        Get if the auto type registration is disabled.
        Returns:
        if the auto type registration is disabled.
        See Also:
        FLIP-131: Consolidate the user-facing Dataflow SDKs/APIs (and deprecate the DataSet API

      • disableAutoTypeRegistration

        @Deprecated
public void disableAutoTypeRegistration()
        Deprecated.
        The method is deprecated because it's only used in DataSet API. All Flink DataSet APIs are deprecated since Flink 1.18 and will be removed in a future Flink major version. You can still build your application in DataSet, but you should move to either the DataStream and/or Table API.
        Control whether Flink is automatically registering all types in the user programs with Kryo.
        See Also:
        FLIP-131: Consolidate the user-facing Dataflow SDKs/APIs (and deprecate the DataSet API

      • isUseSnapshotCompression

        public boolean isUseSnapshotCompression()

      • setUseSnapshotCompression

        public void setUseSnapshotCompression​(boolean useSnapshotCompression)

      • getAsyncInflightRecordsLimit

        @Experimental
public int getAsyncInflightRecordsLimit()

      • setAsyncInflightRecordsLimit

        @Experimental
public ExecutionConfig setAsyncInflightRecordsLimit​(int limit)

      • getAsyncStateBufferSize

        @Experimental
public int getAsyncStateBufferSize()

      • setAsyncStateBufferSize

        @Experimental
public ExecutionConfig setAsyncStateBufferSize​(int bufferSize)

      • getAsyncStateBufferTimeout

        @Experimental
public long getAsyncStateBufferTimeout()

      • setAsyncStateBufferTimeout

        @Experimental
public ExecutionConfig setAsyncStateBufferTimeout​(long timeout)

      • hashCode

        public int hashCode()
        Overrides:
        hashCode in class Object

      • canEqual

        @Deprecated
public boolean canEqual​(Object obj)
        Deprecated.
        It is not intended to be used by users.
        This method simply checks whether the object is an ExecutionConfig instance.

      • configure

        public void configure​(ReadableConfig configuration,
                      ClassLoader classLoader)
        Sets all relevant options contained in the ReadableConfig such as e.g. PipelineOptions.CLOSURE_CLEANER_LEVEL.

        It will change the value of a setting only if a corresponding option was set in the configuration. If a key is not present, the current value of a field will remain untouched.

        Parameters:
        configuration - a configuration to read the values from
        classLoader - a class loader to use when loading classes

      • toConfiguration

        @Internal
public Configuration toConfiguration()
        Returns:
        A copy of internal configuration. Note it is missing all options that are stored as plain java fields in ExecutionConfig.