You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2024/10/03 13:44:32 UTC
(flink) 02/03: [FLINK-19153] Remove deprecated ExecutionMode in flink-core module
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8931ce4494ee730126cb856c8a3a8c59a1d5dcf6
Author: codenohup <hu...@gmail.com>
AuthorDate: Tue Oct 1 19:46:11 2024 +0800
[FLINK-19153] Remove deprecated ExecutionMode in flink-core module
---
.../apache/flink/api/common/ExecutionConfig.java | 60 ----------
.../org/apache/flink/api/common/ExecutionMode.java | 92 ---------------
.../docs/reference/pyflink.common/config.rst | 11 --
flink-python/pyflink/common/__init__.py | 2 -
flink-python/pyflink/common/execution_config.py | 39 -------
flink-python/pyflink/common/execution_mode.py | 82 -------------
.../pyflink/common/tests/test_execution_config.py | 20 +---
.../flink/runtime/io/network/DataExchangeMode.java | 127 ---------------------
.../runtime/io/network/DataExchangeModeTest.java | 50 --------
.../scheduler/benchmark/JobConfiguration.java | 48 ++++----
.../benchmark/SchedulerBenchmarkUtils.java | 4 +-
.../streaming/api/datastream/DataStreamUtils.java | 38 ------
.../recovery/BatchFineGrainedRecoveryITCase.java | 5 +-
13 files changed, 27 insertions(+), 551 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java b/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
index fc8c3480c24..9dd729cae5b 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
@@ -24,7 +24,6 @@ import org.apache.flink.annotation.Public;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.serialization.SerializerConfig;
import org.apache.flink.api.common.serialization.SerializerConfigImpl;
-import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.DescribedEnum;
@@ -48,7 +47,6 @@ import java.util.Map;
import java.util.Objects;
import java.util.Optional;
-import static org.apache.flink.configuration.ConfigOptions.key;
import static org.apache.flink.configuration.description.TextElement.text;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -61,8 +59,6 @@ import static org.apache.flink.util.Preconditions.checkArgument;
* functions that do not define a specific value directly.
* <li>The number of retries in the case of failed executions.
* <li>The delay between execution retries.
- * <li>The {@link ExecutionMode} of the program: Batch or Pipelined. The default execution mode is
- * {@link ExecutionMode#PIPELINED}
* <li>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
@@ -104,22 +100,6 @@ public class ExecutionConfig implements Serializable, Archiveable<ArchivedExecut
*/
public static final int PARALLELISM_UNKNOWN = -2;
- /**
- * Internal {@link ConfigOption}s, that are not exposed and it's not possible to configure them
- * via config files. We are defining them here, so that we can store them in the {@link
- * #configuration}.
- *
- * <p>If you decide to expose any of those {@link ConfigOption}s, please double-check if the
- * key, type and descriptions are sensible, as the initial values are arbitrary.
- */
- // --------------------------------------------------------------------------------------------
-
- private static final ConfigOption<ExecutionMode> EXECUTION_MODE =
- key("hidden.execution.mode")
- .enumType(ExecutionMode.class)
- .defaultValue(ExecutionMode.PIPELINED)
- .withDescription("Defines how data exchange happens - batch or pipelined");
-
// --------------------------------------------------------------------------------------------
/**
@@ -408,46 +388,6 @@ public class ExecutionConfig implements Serializable, Archiveable<ArchivedExecut
return configuration.getOptional(JobManagerOptions.SCHEDULER);
}
- /**
- * 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.
- *
- * <p>The default execution mode is {@link ExecutionMode#PIPELINED}.
- *
- * @param executionMode The execution mode to use.
- * @deprecated The {@link 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.
- * @see <a href="https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=158866741">
- * FLIP-131: Consolidate the user-facing Dataflow SDKs/APIs (and deprecate the DataSet
- * API</a>
- */
- @Deprecated
- public void setExecutionMode(ExecutionMode executionMode) {
- configuration.set(EXECUTION_MODE, executionMode);
- }
-
- /**
- * Gets the execution mode used to execute the program. The execution mode defines whether data
- * exchanges are performed in a batch or on a pipelined manner.
- *
- * <p>The default execution mode is {@link ExecutionMode#PIPELINED}.
- *
- * @return The execution mode for the program.
- * @deprecated The {@link 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.
- * @see <a href="https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=158866741">
- * FLIP-131: Consolidate the user-facing Dataflow SDKs/APIs (and deprecate the DataSet
- * API</a>
- */
- @Deprecated
- public ExecutionMode getExecutionMode() {
- return configuration.get(EXECUTION_MODE);
- }
-
/**
* Enables the Flink runtime to auto-generate UID's for operators.
*
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionMode.java b/flink-core/src/main/java/org/apache/flink/api/common/ExecutionMode.java
deleted file mode 100644
index e60aad819aa..00000000000
--- a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionMode.java
+++ /dev/null
@@ -1,92 +0,0 @@
-/*
- * 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.api.common;
-
-import org.apache.flink.annotation.Public;
-
-/**
- * The execution mode specifies how a batch program is executed in terms of data exchange:
- * pipelining or batched.
- *
- * @deprecated The {@link 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.
- * @see <a href="https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=158866741">
- * FLIP-131: Consolidate the user-facing Dataflow SDKs/APIs (and deprecate the DataSet API</a>
- */
-@Deprecated
-@Public
-public enum ExecutionMode {
-
- /**
- * Executes the program in a pipelined fashion (including shuffles and broadcasts), except for
- * data exchanges that are susceptible to deadlocks when pipelining. These data exchanges are
- * performed in a batch manner.
- *
- * <p>An example of situations that are susceptible to deadlocks (when executed in a pipelined
- * manner) are data flows that branch (one data set consumed by multiple operations) and re-join
- * later:
- *
- * <pre>{@code
- * DataSet data = ...;
- * DataSet mapped1 = data.map(new MyMapper());
- * DataSet mapped2 = data.map(new AnotherMapper());
- * mapped1.join(mapped2).where(...).equalTo(...);
- * }</pre>
- */
- PIPELINED,
-
- /**
- * Executes the program in a pipelined fashion (including shuffles and broadcasts),
- * <strong>including</strong> data exchanges that are susceptible to deadlocks when executed via
- * pipelining.
- *
- * <p>Usually, {@link #PIPELINED} is the preferable option, which pipelines most data exchanges
- * and only uses batch data exchanges in situations that are susceptible to deadlocks.
- *
- * <p>This option should only be used with care and only in situations where the programmer is
- * sure that the program is safe for full pipelining and that Flink was too conservative when
- * choosing the batch exchange at a certain point.
- */
- PIPELINED_FORCED,
-
- // This is for later, we are missing a bit of infrastructure for this.
- // /**
- // * The execution mode starts executing the program in a pipelined fashion
- // * (except for deadlock prone situations), similar to the {@link #PIPELINED}
- // * option. In the case of a task failure, re-execution happens in a batched
- // * mode, as defined for the {@link #BATCH} option.
- // */
- // PIPELINED_WITH_BATCH_FALLBACK,
-
- /**
- * This mode executes all shuffles and broadcasts in a batch fashion, while pipelining data
- * between operations that exchange data only locally between one producer and one consumer.
- */
- BATCH,
-
- /**
- * This mode executes the program in a strict batch way, including all points where data is
- * forwarded locally from one producer to one consumer. This mode is typically more expensive to
- * execute than the {@link #BATCH} mode. It does guarantee that no successive operations are
- * ever executed concurrently.
- */
- BATCH_FORCED
-}
diff --git a/flink-python/docs/reference/pyflink.common/config.rst b/flink-python/docs/reference/pyflink.common/config.rst
index fa9611e409a..144cfcd938c 100644
--- a/flink-python/docs/reference/pyflink.common/config.rst
+++ b/flink-python/docs/reference/pyflink.common/config.rst
@@ -51,17 +51,6 @@ ExecutionConfig
ExecutionConfig
-ExecutionMode
--------------
-
-.. currentmodule:: pyflink.common.execution_mode
-
-.. autosummary::
- :toctree: api/
-
- ExecutionMode
-
-
RestartStrategy
---------------
diff --git a/flink-python/pyflink/common/__init__.py b/flink-python/pyflink/common/__init__.py
index 19f76febc7d..82328522561 100644
--- a/flink-python/pyflink/common/__init__.py
+++ b/flink-python/pyflink/common/__init__.py
@@ -52,7 +52,6 @@ from pyflink.common.completable_future import CompletableFuture
from pyflink.common.config_options import ConfigOption, ConfigOptions
from pyflink.common.configuration import Configuration
from pyflink.common.execution_config import ExecutionConfig
-from pyflink.common.execution_mode import ExecutionMode
from pyflink.common.input_dependency_constraint import InputDependencyConstraint
from pyflink.common.job_client import JobClient
from pyflink.common.job_execution_result import JobExecutionResult
@@ -81,7 +80,6 @@ __all__ = [
'SimpleStringSchema',
'Encoder',
'CompletableFuture',
- 'ExecutionMode',
'InputDependencyConstraint',
'JobClient',
'JobExecutionResult',
diff --git a/flink-python/pyflink/common/execution_config.py b/flink-python/pyflink/common/execution_config.py
index db46d957551..efdf0c2de8d 100644
--- a/flink-python/pyflink/common/execution_config.py
+++ b/flink-python/pyflink/common/execution_config.py
@@ -18,7 +18,6 @@
from typing import Dict, List
-from pyflink.common.execution_mode import ExecutionMode
from pyflink.java_gateway import get_gateway
__all__ = ['ExecutionConfig']
@@ -36,9 +35,6 @@ class ExecutionConfig(object):
- The delay between execution retries.
- - The :class:`ExecutionMode` of the program: Batch or Pipelined.
- The default execution mode is :data:`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
@@ -246,41 +242,6 @@ class ExecutionConfig(object):
self._j_execution_config = self._j_execution_config.setTaskCancellationTimeout(timeout)
return self
- def set_execution_mode(self, execution_mode: ExecutionMode) -> 'ExecutionConfig':
- """
- 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 :data:`ExecutionMode.PIPELINED`.
-
- Example:
- ::
-
- >>> config.set_execution_mode(ExecutionMode.BATCH)
-
- :param execution_mode: The execution mode to use. The execution mode could be
- :data:`ExecutionMode.PIPELINED`,
- :data:`ExecutionMode.PIPELINED_FORCED`,
- :data:`ExecutionMode.BATCH` or
- :data:`ExecutionMode.BATCH_FORCED`.
- """
- self._j_execution_config.setExecutionMode(execution_mode._to_j_execution_mode())
- return self
-
- def get_execution_mode(self) -> 'ExecutionMode':
- """
- Gets the execution mode used 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 :data:`ExecutionMode.PIPELINED`.
-
- .. seealso:: :func:`set_execution_mode`
-
- :return: The execution mode for the program.
- """
- j_execution_mode = self._j_execution_config.getExecutionMode()
- return ExecutionMode._from_j_execution_mode(j_execution_mode)
-
def enable_force_kryo(self) -> 'ExecutionConfig':
"""
Force TypeExtractor to use Kryo serializer for POJOS even though we could analyze as POJO.
diff --git a/flink-python/pyflink/common/execution_mode.py b/flink-python/pyflink/common/execution_mode.py
deleted file mode 100644
index 5252024526c..00000000000
--- a/flink-python/pyflink/common/execution_mode.py
+++ /dev/null
@@ -1,82 +0,0 @@
-################################################################################
-# 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.
-################################################################################
-from enum import Enum
-
-from pyflink.java_gateway import get_gateway
-
-__all__ = ['ExecutionMode']
-
-
-class ExecutionMode(Enum):
- """
- The execution mode specifies how a batch program is executed in terms
- of data exchange: pipelining or batched.
-
- :data:`PIPELINED`:
-
- Executes the program in a pipelined fashion (including shuffles and broadcasts),
- except for data exchanges that are susceptible to deadlocks when pipelining.
- These data exchanges are performed in a batch manner.
-
- An example of situations that are susceptible to deadlocks (when executed in a
- pipelined manner) are data flows that branch (one data set consumed by multiple
- operations) and re-join later.
-
-
- :data:`PIPELINED_FORCED`:
-
- Executes the program in a pipelined fashion (including shuffles and broadcasts),
- **including** data exchanges that are susceptible to deadlocks when
- executed via pipelining.
-
- Usually, PIPELINED is the preferable option, which pipelines most
- data exchanges and only uses batch data exchanges in situations that are
- susceptible to deadlocks.
-
- This option should only be used with care and only in situations where the
- programmer is sure that the program is safe for full pipelining and that
- Flink was too conservative when choosing the batch exchange at a certain
- point.
-
- :data:`BATCH`:
-
- This mode executes all shuffles and broadcasts in a batch fashion, while
- pipelining data between operations that exchange data only locally
- between one producer and one consumer.
-
- :data:`BATCH_FORCED`:
-
- This mode executes the program in a strict batch way, including all points
- where data is forwarded locally from one producer to one consumer. This mode
- is typically more expensive to execute than the BATCH mode. It does
- guarantee that no successive operations are ever executed concurrently.
- """
-
- PIPELINED = 0
- PIPELINED_FORCED = 1
- BATCH = 2
- BATCH_FORCED = 3
-
- @staticmethod
- def _from_j_execution_mode(j_execution_mode) -> 'ExecutionMode':
- return ExecutionMode[j_execution_mode.name()]
-
- def _to_j_execution_mode(self):
- gateway = get_gateway()
- JExecutionMode = gateway.jvm.org.apache.flink.api.common.ExecutionMode
- return getattr(JExecutionMode, self.name)
diff --git a/flink-python/pyflink/common/tests/test_execution_config.py b/flink-python/pyflink/common/tests/test_execution_config.py
index e08e5a23e83..3a8c278959f 100644
--- a/flink-python/pyflink/common/tests/test_execution_config.py
+++ b/flink-python/pyflink/common/tests/test_execution_config.py
@@ -16,7 +16,7 @@
# limitations under the License.
################################################################################
from pyflink.datastream import StreamExecutionEnvironment
-from pyflink.common import (ExecutionConfig, ExecutionMode, Configuration)
+from pyflink.common import (ExecutionConfig, Configuration)
from pyflink.java_gateway import get_gateway
from pyflink.testing.test_case_utils import PyFlinkTestCase
from pyflink.util.java_utils import get_j_env_configuration
@@ -91,24 +91,6 @@ class ExecutionConfigTests(PyFlinkTestCase):
self.assertEqual(self.execution_config.get_task_cancellation_timeout(), 3000)
- def test_get_set_execution_mode(self):
-
- self.execution_config.set_execution_mode(ExecutionMode.BATCH)
-
- self.assertEqual(self.execution_config.get_execution_mode(), ExecutionMode.BATCH)
-
- self.execution_config.set_execution_mode(ExecutionMode.PIPELINED)
-
- self.assertEqual(self.execution_config.get_execution_mode(), ExecutionMode.PIPELINED)
-
- self.execution_config.set_execution_mode(ExecutionMode.BATCH_FORCED)
-
- self.assertEqual(self.execution_config.get_execution_mode(), ExecutionMode.BATCH_FORCED)
-
- self.execution_config.set_execution_mode(ExecutionMode.PIPELINED_FORCED)
-
- self.assertEqual(self.execution_config.get_execution_mode(), ExecutionMode.PIPELINED_FORCED)
-
def test_disable_enable_auto_generated_uids(self):
self.execution_config.disable_auto_generated_uids()
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/DataExchangeMode.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/DataExchangeMode.java
deleted file mode 100644
index 74e63f760ba..00000000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/DataExchangeMode.java
+++ /dev/null
@@ -1,127 +0,0 @@
-/*
- * 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.runtime.io.network;
-
-import org.apache.flink.api.common.ExecutionMode;
-import org.apache.flink.runtime.operators.shipping.ShipStrategyType;
-
-/** Defines how the data exchange between two specific operators happens. */
-public enum DataExchangeMode {
-
- /**
- * The data exchange is streamed, sender and receiver are online at the same time, and the
- * receiver back-pressures the sender.
- */
- PIPELINED,
-
- /**
- * The data exchange is decoupled. The sender first produces its entire result and finishes.
- * After that, the receiver is started and may consume the data.
- */
- BATCH,
-
- /**
- * The data exchange starts like in {@link #PIPELINED} and falls back to {@link #BATCH} for
- * recovery runs.
- */
- PIPELINE_WITH_BATCH_FALLBACK;
-
- // ------------------------------------------------------------------------
-
- public static DataExchangeMode getForForwardExchange(ExecutionMode mode) {
- return FORWARD[mode.ordinal()];
- }
-
- public static DataExchangeMode getForShuffleOrBroadcast(ExecutionMode mode) {
- return SHUFFLE[mode.ordinal()];
- }
-
- public static DataExchangeMode getPipelineBreakingExchange(ExecutionMode mode) {
- return BREAKING[mode.ordinal()];
- }
-
- /**
- * Computes the mode of data exchange to be used for a given execution mode and ship strategy.
- * The type of the data exchange depends also on whether this connection has been identified to
- * require pipeline breaking for deadlock avoidance.
- *
- * <ul>
- * <li>If the connection is set to be pipeline breaking, this returns the pipeline breaking
- * variant of the execution mode {@link
- * org.apache.flink.runtime.io.network.DataExchangeMode#getPipelineBreakingExchange(org.apache.flink.api.common.ExecutionMode)}.
- * <li>If the data exchange is a simple FORWARD (one-to-one communication), this returns
- * {@link
- * org.apache.flink.runtime.io.network.DataExchangeMode#getForForwardExchange(org.apache.flink.api.common.ExecutionMode)}.
- * <li>If otherwise, this returns {@link
- * org.apache.flink.runtime.io.network.DataExchangeMode#getForShuffleOrBroadcast(org.apache.flink.api.common.ExecutionMode)}.
- * </ul>
- *
- * @param shipStrategy The ship strategy (FORWARD, PARTITION, BROADCAST, ...) of the runtime
- * data exchange.
- * @return The data exchange mode for the connection, given the concrete ship strategy.
- */
- public static DataExchangeMode select(
- ExecutionMode executionMode, ShipStrategyType shipStrategy, boolean breakPipeline) {
-
- if (shipStrategy == null || shipStrategy == ShipStrategyType.NONE) {
- throw new IllegalArgumentException("shipStrategy may not be null or NONE");
- }
- if (executionMode == null) {
- throw new IllegalArgumentException("executionMode may not mbe null");
- }
-
- if (breakPipeline) {
- return getPipelineBreakingExchange(executionMode);
- } else if (shipStrategy == ShipStrategyType.FORWARD) {
- return getForForwardExchange(executionMode);
- } else {
- return getForShuffleOrBroadcast(executionMode);
- }
- }
-
- // ------------------------------------------------------------------------
-
- private static final DataExchangeMode[] FORWARD =
- new DataExchangeMode[ExecutionMode.values().length];
-
- private static final DataExchangeMode[] SHUFFLE =
- new DataExchangeMode[ExecutionMode.values().length];
-
- private static final DataExchangeMode[] BREAKING =
- new DataExchangeMode[ExecutionMode.values().length];
-
- // initialize the map between execution modes and exchange modes in
- static {
- FORWARD[ExecutionMode.PIPELINED_FORCED.ordinal()] = PIPELINED;
- SHUFFLE[ExecutionMode.PIPELINED_FORCED.ordinal()] = PIPELINED;
- BREAKING[ExecutionMode.PIPELINED_FORCED.ordinal()] = PIPELINED;
-
- FORWARD[ExecutionMode.PIPELINED.ordinal()] = PIPELINED;
- SHUFFLE[ExecutionMode.PIPELINED.ordinal()] = PIPELINED;
- BREAKING[ExecutionMode.PIPELINED.ordinal()] = BATCH;
-
- FORWARD[ExecutionMode.BATCH.ordinal()] = PIPELINED;
- SHUFFLE[ExecutionMode.BATCH.ordinal()] = BATCH;
- BREAKING[ExecutionMode.BATCH.ordinal()] = BATCH;
-
- FORWARD[ExecutionMode.BATCH_FORCED.ordinal()] = BATCH;
- SHUFFLE[ExecutionMode.BATCH_FORCED.ordinal()] = BATCH;
- BREAKING[ExecutionMode.BATCH_FORCED.ordinal()] = BATCH;
- }
-}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/DataExchangeModeTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/DataExchangeModeTest.java
deleted file mode 100644
index da5506c78c4..00000000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/DataExchangeModeTest.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * 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.runtime.io.network;
-
-import org.apache.flink.api.common.ExecutionMode;
-
-import org.junit.jupiter.api.Test;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** This test verifies that the data exchange modes are defined for every execution mode. */
-class DataExchangeModeTest {
-
- @Test
- void testForward() {
- for (ExecutionMode mode : ExecutionMode.values()) {
- assertThat(DataExchangeMode.getForForwardExchange(mode)).isNotNull();
- }
- }
-
- @Test
- void testShuffleAndBroadcast() {
- for (ExecutionMode mode : ExecutionMode.values()) {
- assertThat(DataExchangeMode.getForShuffleOrBroadcast(mode)).isNotNull();
- }
- }
-
- @Test
- void testPipelineBreaking() {
- for (ExecutionMode mode : ExecutionMode.values()) {
- assertThat(DataExchangeMode.getPipelineBreakingExchange(mode)).isNotNull();
- }
- }
-}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/JobConfiguration.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/JobConfiguration.java
index 5a049d98172..20ee4d94300 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/JobConfiguration.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/JobConfiguration.java
@@ -18,7 +18,7 @@
package org.apache.flink.runtime.scheduler.benchmark;
-import org.apache.flink.api.common.ExecutionMode;
+import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.JobManagerOptions.HybridPartitionDataConsumeConstraint;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
@@ -26,15 +26,15 @@ import org.apache.flink.runtime.jobgraph.JobType;
/**
* {@link JobConfiguration} contains the configuration of a STREAMING/BATCH job. It concludes {@link
- * DistributionPattern}, {@link ResultPartitionType}, {@link JobType}, {@link ExecutionMode}, {@link
- * HybridPartitionDataConsumeConstraint}.
+ * DistributionPattern}, {@link ResultPartitionType}, {@link JobType}, {@link RuntimeExecutionMode},
+ * {@link HybridPartitionDataConsumeConstraint}.
*/
public enum JobConfiguration {
STREAMING(
DistributionPattern.ALL_TO_ALL,
ResultPartitionType.PIPELINED,
JobType.STREAMING,
- ExecutionMode.PIPELINED,
+ RuntimeExecutionMode.STREAMING,
4000,
false),
@@ -42,7 +42,7 @@ public enum JobConfiguration {
DistributionPattern.ALL_TO_ALL,
ResultPartitionType.BLOCKING,
JobType.BATCH,
- ExecutionMode.BATCH,
+ RuntimeExecutionMode.BATCH,
4000,
false),
@@ -50,7 +50,7 @@ public enum JobConfiguration {
DistributionPattern.ALL_TO_ALL,
ResultPartitionType.HYBRID_FULL,
JobType.BATCH,
- ExecutionMode.BATCH,
+ RuntimeExecutionMode.BATCH,
HybridPartitionDataConsumeConstraint.UNFINISHED_PRODUCERS,
4000,
false),
@@ -59,7 +59,7 @@ public enum JobConfiguration {
DistributionPattern.ALL_TO_ALL,
ResultPartitionType.HYBRID_FULL,
JobType.BATCH,
- ExecutionMode.BATCH,
+ RuntimeExecutionMode.BATCH,
HybridPartitionDataConsumeConstraint.ONLY_FINISHED_PRODUCERS,
4000,
false),
@@ -68,7 +68,7 @@ public enum JobConfiguration {
DistributionPattern.ALL_TO_ALL,
ResultPartitionType.HYBRID_FULL,
JobType.BATCH,
- ExecutionMode.BATCH,
+ RuntimeExecutionMode.BATCH,
HybridPartitionDataConsumeConstraint.ALL_PRODUCERS_FINISHED,
4000,
false),
@@ -77,7 +77,7 @@ public enum JobConfiguration {
DistributionPattern.ALL_TO_ALL,
ResultPartitionType.PIPELINED,
JobType.STREAMING,
- ExecutionMode.PIPELINED,
+ RuntimeExecutionMode.STREAMING,
10,
false),
@@ -85,7 +85,7 @@ public enum JobConfiguration {
DistributionPattern.ALL_TO_ALL,
ResultPartitionType.BLOCKING,
JobType.BATCH,
- ExecutionMode.BATCH,
+ RuntimeExecutionMode.BATCH,
10,
false),
@@ -93,7 +93,7 @@ public enum JobConfiguration {
DistributionPattern.ALL_TO_ALL,
ResultPartitionType.HYBRID_FULL,
JobType.BATCH,
- ExecutionMode.BATCH,
+ RuntimeExecutionMode.BATCH,
HybridPartitionDataConsumeConstraint.UNFINISHED_PRODUCERS,
10,
false),
@@ -102,7 +102,7 @@ public enum JobConfiguration {
DistributionPattern.ALL_TO_ALL,
ResultPartitionType.HYBRID_FULL,
JobType.BATCH,
- ExecutionMode.BATCH,
+ RuntimeExecutionMode.BATCH,
HybridPartitionDataConsumeConstraint.ONLY_FINISHED_PRODUCERS,
10,
false),
@@ -111,7 +111,7 @@ public enum JobConfiguration {
DistributionPattern.ALL_TO_ALL,
ResultPartitionType.HYBRID_FULL,
JobType.BATCH,
- ExecutionMode.BATCH,
+ RuntimeExecutionMode.BATCH,
HybridPartitionDataConsumeConstraint.ALL_PRODUCERS_FINISHED,
10,
false),
@@ -120,7 +120,7 @@ public enum JobConfiguration {
DistributionPattern.ALL_TO_ALL,
ResultPartitionType.PIPELINED,
JobType.STREAMING,
- ExecutionMode.PIPELINED,
+ RuntimeExecutionMode.STREAMING,
4000,
true),
@@ -128,7 +128,7 @@ public enum JobConfiguration {
DistributionPattern.ALL_TO_ALL,
ResultPartitionType.BLOCKING,
JobType.BATCH,
- ExecutionMode.BATCH,
+ RuntimeExecutionMode.BATCH,
4000,
true),
@@ -136,7 +136,7 @@ public enum JobConfiguration {
DistributionPattern.ALL_TO_ALL,
ResultPartitionType.PIPELINED,
JobType.STREAMING,
- ExecutionMode.PIPELINED,
+ RuntimeExecutionMode.STREAMING,
10,
true),
@@ -144,7 +144,7 @@ public enum JobConfiguration {
DistributionPattern.ALL_TO_ALL,
ResultPartitionType.BLOCKING,
JobType.BATCH,
- ExecutionMode.BATCH,
+ RuntimeExecutionMode.STREAMING,
10,
true);
@@ -152,7 +152,7 @@ public enum JobConfiguration {
private final DistributionPattern distributionPattern;
private final ResultPartitionType resultPartitionType;
private final JobType jobType;
- private final ExecutionMode executionMode;
+ private final RuntimeExecutionMode runtimeExecutionMode;
private final boolean evenlySpreadOutSlots;
private final HybridPartitionDataConsumeConstraint hybridPartitionDataConsumeConstraint;
@@ -160,14 +160,14 @@ public enum JobConfiguration {
DistributionPattern distributionPattern,
ResultPartitionType resultPartitionType,
JobType jobType,
- ExecutionMode executionMode,
+ RuntimeExecutionMode runtimeExecutionMode,
int parallelism,
boolean evenlySpreadOutSlots) {
this(
distributionPattern,
resultPartitionType,
jobType,
- executionMode,
+ runtimeExecutionMode,
HybridPartitionDataConsumeConstraint.UNFINISHED_PRODUCERS,
parallelism,
evenlySpreadOutSlots);
@@ -177,14 +177,14 @@ public enum JobConfiguration {
DistributionPattern distributionPattern,
ResultPartitionType resultPartitionType,
JobType jobType,
- ExecutionMode executionMode,
+ RuntimeExecutionMode runtimeExecutionMode,
HybridPartitionDataConsumeConstraint hybridPartitionDataConsumeConstraint,
int parallelism,
boolean evenlySpreadOutSlots) {
this.distributionPattern = distributionPattern;
this.resultPartitionType = resultPartitionType;
this.jobType = jobType;
- this.executionMode = executionMode;
+ this.runtimeExecutionMode = runtimeExecutionMode;
this.hybridPartitionDataConsumeConstraint = hybridPartitionDataConsumeConstraint;
this.parallelism = parallelism;
this.evenlySpreadOutSlots = evenlySpreadOutSlots;
@@ -206,8 +206,8 @@ public enum JobConfiguration {
return jobType;
}
- public ExecutionMode getExecutionMode() {
- return executionMode;
+ public RuntimeExecutionMode getRuntimeExecutionMode() {
+ return runtimeExecutionMode;
}
public HybridPartitionDataConsumeConstraint getHybridPartitionDataConsumeConstraint() {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/SchedulerBenchmarkUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/SchedulerBenchmarkUtils.java
index a60b3fd584b..af718fdd7cf 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/SchedulerBenchmarkUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/SchedulerBenchmarkUtils.java
@@ -95,9 +95,7 @@ public class SchedulerBenchmarkUtils {
jobGraph.setJobType(jobConfiguration.getJobType());
- final ExecutionConfig executionConfig = new ExecutionConfig();
- executionConfig.setExecutionMode(jobConfiguration.getExecutionMode());
- jobGraph.setExecutionConfig(executionConfig);
+ jobGraph.setExecutionConfig(new ExecutionConfig());
return jobGraph;
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStreamUtils.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStreamUtils.java
index 6c65053c46e..a35494a7143 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStreamUtils.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStreamUtils.java
@@ -24,48 +24,10 @@ import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.streaming.api.transformations.PartitionTransformation;
import org.apache.flink.streaming.runtime.partitioner.ForwardPartitioner;
-import java.util.Iterator;
-
/** A collection of utilities for {@link DataStream DataStreams}. */
@Experimental
public final class DataStreamUtils {
- /**
- * Triggers the distributed execution of the streaming dataflow and returns an iterator over the
- * elements of the given DataStream.
- *
- * <p>The DataStream application is executed in the regular distributed manner on the target
- * environment, and the events from the stream are polled back to this application process and
- * thread through Flink's REST API.
- *
- * @deprecated Please use {@link DataStream#executeAndCollect()}.
- */
- @Deprecated
- public static <OUT> Iterator<OUT> collect(DataStream<OUT> stream) {
- return collect(stream, "Data Stream Collect");
- }
-
- /**
- * Triggers the distributed execution of the streaming dataflow and returns an iterator over the
- * elements of the given DataStream.
- *
- * <p>The DataStream application is executed in the regular distributed manner on the target
- * environment, and the events from the stream are polled back to this application process and
- * thread through Flink's REST API.
- *
- * @deprecated Please use {@link DataStream#executeAndCollect()}.
- */
- @Deprecated
- public static <OUT> Iterator<OUT> collect(DataStream<OUT> stream, String executionJobName) {
- try {
- return stream.executeAndCollect(executionJobName);
- } catch (Exception e) {
- // this "wrap as unchecked" step is here only to preserve the exception signature
- // backwards compatible.
- throw new RuntimeException("Failed to execute data stream", e);
- }
- }
-
// ------------------------------------------------------------------------
// Deriving a KeyedStream from a stream already partitioned by key
// without a shuffle
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/BatchFineGrainedRecoveryITCase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/BatchFineGrainedRecoveryITCase.java
index b46910033db..1a7843dc88d 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/BatchFineGrainedRecoveryITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/BatchFineGrainedRecoveryITCase.java
@@ -18,7 +18,6 @@
package org.apache.flink.test.recovery;
-import org.apache.flink.api.common.ExecutionMode;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.java.tuple.Tuple2;
@@ -274,9 +273,7 @@ public class BatchFineGrainedRecoveryITCase extends TestLogger {
StreamExecutionEnvironment env = new TestStreamEnvironment(miniCluster, 1);
RestartStrategyUtils.configureFixedDelayRestartStrategy(
env, MAX_JOB_RESTART_ATTEMPTS, Duration.ofMillis(10));
- env.getConfig()
- .setExecutionMode(
- ExecutionMode.BATCH_FORCED); // forces all partitions to be blocking
+ env.setRuntimeMode(RuntimeExecutionMode.BATCH);
return env;
}