You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by zh...@apache.org on 2024/09/12 01:47:28 UTC
(flink) branch master updated (8c6f13cead9 -> 356e2af4b27)
This is an automated email from the ASF dual-hosted git repository.
zhuzh pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 8c6f13cead9 [FLINK-36230][pyflink-table] Fix the incorrect calling of REGEXP_EXTRACT Python Table API
new 581a4f761d2 [FLINK-36185][state-processor] Remove deprecated internal class BoundedOneInputStreamTaskRunner.
new d564e889264 [FLINK-36185][e2e] Rewrite end to end cases with new Kafka source.
new c7b1f4be424 [FLINK-36185][core] Remove RuntimeContext#getExecutionConfig
new 356e2af4b27 [FLINK-36185][e2e] Disable PyFlink end-to-end case that uses Kafka legacy source.
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../flink/api/common/functions/RuntimeContext.java | 12 --
.../functions/util/AbstractRuntimeUDFContext.java | 6 -
.../registry/test/TestAvroConsumerConfluent.java | 33 ++--
.../test-scripts/test_pyflink.sh | 180 +++++++++++----------
.../flink/cep/operator/CepRuntimeContext.java | 7 -
.../output/BoundedOneInputStreamTaskRunner.java | 96 -----------
.../state/api/runtime/SavepointRuntimeContext.java | 7 -
.../util/MockStreamingRuntimeContext.java | 7 +
.../api/functions/async/RichAsyncFunction.java | 7 -
pom.xml | 4 +
10 files changed, 129 insertions(+), 230 deletions(-)
delete mode 100644 flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/BoundedOneInputStreamTaskRunner.java
(flink) 01/04: [FLINK-36185][state-processor] Remove deprecated internal class BoundedOneInputStreamTaskRunner.
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhuzh pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 581a4f761d231166f1471fcd1f18ae55df680983
Author: JunRuiLee <jr...@gmail.com>
AuthorDate: Fri Aug 30 11:42:26 2024 +0800
[FLINK-36185][state-processor] Remove deprecated internal class BoundedOneInputStreamTaskRunner.
---
.../output/BoundedOneInputStreamTaskRunner.java | 96 ----------------------
1 file changed, 96 deletions(-)
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/BoundedOneInputStreamTaskRunner.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/BoundedOneInputStreamTaskRunner.java
deleted file mode 100644
index cefe67f7396..00000000000
--- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/BoundedOneInputStreamTaskRunner.java
+++ /dev/null
@@ -1,96 +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.state.api.output;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.functions.OpenContext;
-import org.apache.flink.api.common.functions.RichMapPartitionFunction;
-import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
-import org.apache.flink.state.api.functions.Timestamper;
-import org.apache.flink.state.api.runtime.SavepointEnvironment;
-import org.apache.flink.streaming.api.graph.StreamConfig;
-import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
-import org.apache.flink.util.Collector;
-
-/**
- * A {@link RichMapPartitionFunction} that serves as the runtime for a {@link BoundedStreamTask}.
- *
- * <p>The task is executed processing the data in a particular partition instead of the pulling from
- * the network stack. After all data has been processed the runner will output the {@link
- * OperatorSubtaskState} from the snapshot of the bounded task.
- *
- * @param <IN> Type of the input to the partition
- */
-@Deprecated
-@Internal
-public class BoundedOneInputStreamTaskRunner<IN>
- extends RichMapPartitionFunction<IN, TaggedOperatorSubtaskState> {
-
- private static final long serialVersionUID = 1L;
-
- private final StreamConfig streamConfig;
-
- private final int maxParallelism;
-
- private final Timestamper<IN> timestamper;
-
- private transient SavepointEnvironment env;
-
- /**
- * Create a new {@link BoundedOneInputStreamTaskRunner}.
- *
- * @param streamConfig The internal configuration for the task.
- * @param maxParallelism The max parallelism of the operator.
- */
- public BoundedOneInputStreamTaskRunner(
- StreamConfig streamConfig, int maxParallelism, Timestamper<IN> timestamper) {
-
- this.streamConfig = streamConfig;
- this.maxParallelism = maxParallelism;
- this.timestamper = timestamper;
- }
-
- @Override
- public void open(OpenContext openContext) throws Exception {
- super.open(openContext);
-
- env =
- new SavepointEnvironment.Builder(
- getRuntimeContext(),
- getRuntimeContext().getExecutionConfig(),
- maxParallelism)
- .setConfiguration(streamConfig.getConfiguration())
- .build();
- }
-
- @Override
- public void mapPartition(Iterable<IN> values, Collector<TaggedOperatorSubtaskState> out)
- throws Exception {
- BoundedStreamTask<
- IN,
- TaggedOperatorSubtaskState,
- ? extends OneInputStreamOperator<IN, TaggedOperatorSubtaskState>>
- boundedStreamTask = new BoundedStreamTask<>(env, values, timestamper, out);
- try {
- boundedStreamTask.invoke();
- } finally {
- boundedStreamTask.cleanUp(null);
- }
- }
-}
(flink) 04/04: [FLINK-36185][e2e] Disable PyFlink end-to-end case that uses Kafka legacy source.
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhuzh pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 356e2af4b27b5a7250d9780ee7b41ce945b98c1e
Author: JunRuiLee <jr...@gmail.com>
AuthorDate: Wed Sep 11 09:55:16 2024 +0800
[FLINK-36185][e2e] Disable PyFlink end-to-end case that uses Kafka legacy source.
---
.../test-scripts/test_pyflink.sh | 180 +++++++++++----------
1 file changed, 98 insertions(+), 82 deletions(-)
diff --git a/flink-end-to-end-tests/test-scripts/test_pyflink.sh b/flink-end-to-end-tests/test-scripts/test_pyflink.sh
index a078e84da6b..9eb688ce2ed 100755
--- a/flink-end-to-end-tests/test-scripts/test_pyflink.sh
+++ b/flink-end-to-end-tests/test-scripts/test_pyflink.sh
@@ -61,7 +61,14 @@ source "${CURRENT_DIR}"/kafka_sql_common.sh \
function test_clean_up {
stop_cluster
- stop_kafka_cluster
+ # Note: The 'data_stream_job.py' still uses the Kafka legacy source,
+ # so we temporarily disable this case because we plan to address this
+ # in future updates. Once we align with the upcoming PyFlink and Flink 2.0
+ # compatibility efforts, we will remove this temporary disable.
+ #
+ # So we should also skip stopping the Kafka cluster because we did not start the Kafka cluster.
+
+ # stop_kafka_cluster
}
on_exit test_clean_up
@@ -191,90 +198,99 @@ JOB_ID=$($FLINK_DIR/bin/sql-client.sh \
wait_job_terminal_state "$JOB_ID" "FINISHED"
-echo "Test PyFlink DataStream job:"
+# Note: The 'data_stream_job.py' still uses the Kafka legacy source,
+# so we temporarily disable this case because we plan to address this
+# in future updates. Once we align with the upcoming PyFlink and Flink 2.0
+# compatibility efforts, we will remove this temporary disable.
-# prepare Kafka
-echo "Preparing Kafka..."
-
-setup_kafka_dist
-
-start_kafka_cluster
-
-# End to end test for DataStream ProcessFunction with timer
-create_kafka_topic 1 1 timer-stream-source
-create_kafka_topic 1 1 timer-stream-sink
-
-PAYMENT_MSGS='{"createTime": 1603679413000, "orderId": 1603679414, "payAmount": 83685.44904332698, "payPlatform": 0, "provinceId": 3}
-{"createTime": 1603679426000, "orderId": 1603679427, "payAmount": 30092.50657757042, "payPlatform": 0, "provinceId": 1}
-{"createTime": 1603679427000, "orderId": 1603679428, "payAmount": 62644.01719293056, "payPlatform": 0, "provinceId": 6}
-{"createTime": 1603679428000, "orderId": 1603679429, "payAmount": 6449.806795118451, "payPlatform": 0, "provinceId": 2}
-{"createTime": 1603679491000, "orderId": 1603679492, "payAmount": 41108.36128417494, "payPlatform": 0, "provinceId": 0}
-{"createTime": 1603679492000, "orderId": 1603679493, "payAmount": 64882.44233197067, "payPlatform": 0, "provinceId": 4}
-{"createTime": 1603679521000, "orderId": 1603679522, "payAmount": 81648.80712644062, "payPlatform": 0, "provinceId": 3}
-{"createTime": 1603679522000, "orderId": 1603679523, "payAmount": 81861.73063103345, "payPlatform": 0, "provinceId": 4}'
-
-function send_msg_to_kafka {
- while read line
- do
- send_messages_to_kafka "$line" "timer-stream-source"
- sleep 1
- done <<< "$1"
-}
-
-function read_msg_from_kafka {
- $KAFKA_DIR/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning \
- --max-messages $1 \
- --topic $2 \
- --consumer-property group.id=$3 --timeout-ms 90000 2> /dev/null
-}
-
-send_msg_to_kafka "${PAYMENT_MSGS[*]}"
-
-JOB_ID=$(${FLINK_DIR}/bin/flink run \
- -pyfs "${FLINK_PYTHON_TEST_DIR}/python/datastream" \
- -pyreq "${REQUIREMENTS_PATH}" \
- -pyarch "${TEST_DATA_DIR}/venv.zip" \
- -pyexec "venv.zip/.conda/bin/python" \
- -pym "data_stream_job" \
- -j "${KAFKA_SQL_JAR}")
-
-echo "${JOB_ID}"
-JOB_ID=`echo "${JOB_ID}" | sed 's/.* //g'`
-
-wait_job_running ${JOB_ID}
-
-echo "Reading kafka messages..."
-READ_MSG=$(read_msg_from_kafka 16 timer-stream-sink pyflink-e2e-test-timer)
+# echo "Test PyFlink DataStream job:"
+#
+# # Prepare Kafka for the DataStream job
+# echo "Preparing Kafka..."
+#
+# # Setup Kafka distribution
+# setup_kafka_dist
+#
+# # Start Kafka cluster
+# start_kafka_cluster
+#
+# # End to end test for DataStream ProcessFunction with timer
+# # Creating necessary Kafka topics for the test
+# create_kafka_topic 1 1 timer-stream-source
+# create_kafka_topic 1 1 timer-stream-sink
+#
+# # Preparing a set of payment messages in JSON format for the test
+# PAYMENT_MSGS='{"createTime": 1603679413000, "orderId": 1603679414, "payAmount": 83685.44904332698, "payPlatform": 0, "provinceId": 3}
+# {"createTime": 1603679426000, "orderId": 1603679427, "payAmount": 30092.50657757042, "payPlatform": 0, "provinceId": 1}
+# {"createTime": 1603679427000, "orderId": 1603679428, "payAmount": 62644.01719293056, "payPlatform": 0, "provinceId": 6}
+# {"createTime": 1603679428000, "orderId": 1603679429, "payAmount": 6449.806795118451, "payPlatform": 0, "provinceId": 2}
+# {"createTime": 1603679491000, "orderId": 1603679492, "payAmount": 41108.36128417494, "payPlatform": 0, "provinceId": 0}
+# {"createTime": 1603679492000, "orderId": 1603679493, "payAmount": 64882.44233197067, "payPlatform": 0, "provinceId": 4}
+# {"createTime": 1603679521000, "orderId": 1603679522, "payAmount": 81648.80712644062, "payPlatform": 0, "provinceId": 3}
+# {"createTime": 1603679522000, "orderId": 1603679523, "payAmount": 81861.73063103345, "payPlatform": 0, "provinceId": 4}'
+
+# function send_msg_to_kafka {
+# while read line
+# do
+# send_messages_to_kafka "$line" "timer-stream-source"
+# sleep 1
+# done <<< "$1"
+# }
+
+# function read_msg_from_kafka {
+# $KAFKA_DIR/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning \
+# --max-messages $1 \
+# --topic $2 \
+# --consumer-property group.id=$3 --timeout-ms 90000 2> /dev/null
+# }
+
+# send_msg_to_kafka "${PAYMENT_MSGS[*]}"
+
+# JOB_ID=$(${FLINK_DIR}/bin/flink run \
+# -pyfs "${FLINK_PYTHON_TEST_DIR}/python/datastream" \
+# -pyreq "${REQUIREMENTS_PATH}" \
+# -pyarch "${TEST_DATA_DIR}/venv.zip" \
+# -pyexec "venv.zip/.conda/bin/python" \
+# -pym "data_stream_job" \
+# -j "${KAFKA_SQL_JAR}")
+
+# echo "${JOB_ID}"
+# JOB_ID=`echo "${JOB_ID}" | sed 's/.* //g'`
+
+# wait_job_running ${JOB_ID}
+
+# echo "Reading kafka messages..."
+# READ_MSG=$(read_msg_from_kafka 16 timer-stream-sink pyflink-e2e-test-timer)
# We use env.execute_async() to submit the job, cancel it after fetched results.
-cancel_job "${JOB_ID}"
-
-EXPECTED_MSG='Current key: 1603679414, orderId: 1603679414, payAmount: 83685.44904332698, timestamp: 1603679413000
-Current key: 1603679427, orderId: 1603679427, payAmount: 30092.50657757042, timestamp: 1603679426000
-Current key: 1603679428, orderId: 1603679428, payAmount: 62644.01719293056, timestamp: 1603679427000
-Current key: 1603679429, orderId: 1603679429, payAmount: 6449.806795118451, timestamp: 1603679428000
-Current key: 1603679492, orderId: 1603679492, payAmount: 41108.36128417494, timestamp: 1603679491000
-Current key: 1603679493, orderId: 1603679493, payAmount: 64882.44233197067, timestamp: 1603679492000
-Current key: 1603679522, orderId: 1603679522, payAmount: 81648.80712644062, timestamp: 1603679521000
-Current key: 1603679523, orderId: 1603679523, payAmount: 81861.73063103345, timestamp: 1603679522000
-On timer timestamp: -9223372036854774308
-On timer timestamp: -9223372036854774308
-On timer timestamp: -9223372036854774308
-On timer timestamp: -9223372036854774308
-On timer timestamp: -9223372036854774308
-On timer timestamp: -9223372036854774308
-On timer timestamp: -9223372036854774308
-On timer timestamp: -9223372036854774308'
-
-EXPECTED_MSG=$(sort_msg "${EXPECTED_MSG[*]}")
-SORTED_READ_MSG=$(sort_msg "${READ_MSG[*]}")
-
-if [[ "${EXPECTED_MSG[*]}" != "${SORTED_READ_MSG[*]}" ]]; then
- echo "Output from Flink program does not match expected output."
- echo -e "EXPECTED Output: --${EXPECTED_MSG[*]}--"
- echo -e "ACTUAL: --${SORTED_READ_MSG[*]}--"
- exit 1
-fi
+# cancel_job "${JOB_ID}"
+
+# EXPECTED_MSG='Current key: 1603679414, orderId: 1603679414, payAmount: 83685.44904332698, timestamp: 1603679413000
+# Current key: 1603679427, orderId: 1603679427, payAmount: 30092.50657757042, timestamp: 1603679426000
+# Current key: 1603679428, orderId: 1603679428, payAmount: 62644.01719293056, timestamp: 1603679427000
+# Current key: 1603679429, orderId: 1603679429, payAmount: 6449.806795118451, timestamp: 1603679428000
+# Current key: 1603679492, orderId: 1603679492, payAmount: 41108.36128417494, timestamp: 1603679491000
+# Current key: 1603679493, orderId: 1603679493, payAmount: 64882.44233197067, timestamp: 1603679492000
+# Current key: 1603679522, orderId: 1603679522, payAmount: 81648.80712644062, timestamp: 1603679521000
+# Current key: 1603679523, orderId: 1603679523, payAmount: 81861.73063103345, timestamp: 1603679522000
+# On timer timestamp: -9223372036854774308
+# On timer timestamp: -9223372036854774308
+# On timer timestamp: -9223372036854774308
+# On timer timestamp: -9223372036854774308
+# On timer timestamp: -9223372036854774308
+# On timer timestamp: -9223372036854774308
+# On timer timestamp: -9223372036854774308
+# On timer timestamp: -9223372036854774308'
+
+# EXPECTED_MSG=$(sort_msg "${EXPECTED_MSG[*]}")
+# SORTED_READ_MSG=$(sort_msg "${READ_MSG[*]}")
+
+# if [[ "${EXPECTED_MSG[*]}" != "${SORTED_READ_MSG[*]}" ]]; then
+# echo "Output from Flink program does not match expected output."
+# echo -e "EXPECTED Output: --${EXPECTED_MSG[*]}--"
+# echo -e "ACTUAL: --${SORTED_READ_MSG[*]}--"
+# exit 1
+# fi
# clean up python env
"${FLINK_PYTHON_DIR}/dev/lint-python.sh" -r
(flink) 02/04: [FLINK-36185][e2e] Rewrite end to end cases with new Kafka source.
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhuzh pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d564e88926453cfa05863255031999da155b833b
Author: JunRuiLee <jr...@gmail.com>
AuthorDate: Wed Sep 4 13:37:15 2024 +0800
[FLINK-36185][e2e] Rewrite end to end cases with new Kafka source.
Co-authored-by: Qingsheng Ren <re...@gmail.com>
---
.../registry/test/TestAvroConsumerConfluent.java | 33 +++++++++++++---------
1 file changed, 20 insertions(+), 13 deletions(-)
diff --git a/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/java/org/apache/flink/schema/registry/test/TestAvroConsumerConfluent.java b/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/java/org/apache/flink/schema/registry/test/TestAvroConsumerConfluent.java
index e3f41497899..abf5f7104ad 100644
--- a/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/java/org/apache/flink/schema/registry/test/TestAvroConsumerConfluent.java
+++ b/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/java/org/apache/flink/schema/registry/test/TestAvroConsumerConfluent.java
@@ -17,21 +17,23 @@
package org.apache.flink.schema.registry.test;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
+import org.apache.flink.connector.kafka.source.KafkaSource;
+import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
+import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
import org.apache.flink.formats.avro.registry.confluent.ConfluentRegistryAvroDeserializationSchema;
import org.apache.flink.formats.avro.registry.confluent.ConfluentRegistryAvroSerializationSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import example.avro.User;
import org.apache.avro.specific.SpecificRecordBase;
-import org.apache.kafka.clients.producer.ProducerConfig;
import java.util.Properties;
@@ -64,22 +66,28 @@ public class TestAvroConsumerConfluent {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- DataStreamSource<User> input =
- env.addSource(
- new FlinkKafkaConsumer<>(
- parameterTool.getRequired("input-topic"),
+ String bootstrapServers = parameterTool.getRequired("bootstrap.servers");
+ KafkaSource<User> kafkaSource =
+ KafkaSource.<User>builder()
+ .setBootstrapServers(bootstrapServers)
+ .setGroupId(parameterTool.getRequired("group.id"))
+ .setTopics(parameterTool.getRequired("input-topic"))
+ .setDeserializer(
+ KafkaRecordDeserializationSchema.valueOnly(
ConfluentRegistryAvroDeserializationSchema.forSpecific(
- User.class, schemaRegistryUrl),
- config)
- .setStartFromEarliest());
+ User.class, schemaRegistryUrl)))
+ .setStartingOffsets(OffsetsInitializer.earliest())
+ .build();
+
+ DataStreamSource<User> input =
+ env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source");
SingleOutputStreamOperator<String> mapToString =
input.map((MapFunction<User, String>) SpecificRecordBase::toString);
KafkaSink<String> stringSink =
KafkaSink.<String>builder()
- .setBootstrapServers(
- config.getProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))
+ .setBootstrapServers(bootstrapServers)
.setRecordSerializer(
KafkaRecordSerializationSchema.builder()
.setValueSerializationSchema(new SimpleStringSchema())
@@ -91,8 +99,7 @@ public class TestAvroConsumerConfluent {
KafkaSink<User> avroSink =
KafkaSink.<User>builder()
- .setBootstrapServers(
- config.getProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))
+ .setBootstrapServers(bootstrapServers)
.setRecordSerializer(
KafkaRecordSerializationSchema.builder()
.setValueSerializationSchema(
(flink) 03/04: [FLINK-36185][core] Remove RuntimeContext#getExecutionConfig
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhuzh pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c7b1f4be424396bb07cc65777fdb31053f785be1
Author: JunRuiLee <jr...@gmail.com>
AuthorDate: Fri Aug 30 13:41:41 2024 +0800
[FLINK-36185][core] Remove RuntimeContext#getExecutionConfig
This closes #25270.
---
.../apache/flink/api/common/functions/RuntimeContext.java | 12 ------------
.../api/common/functions/util/AbstractRuntimeUDFContext.java | 6 ------
.../org/apache/flink/cep/operator/CepRuntimeContext.java | 7 -------
.../flink/state/api/runtime/SavepointRuntimeContext.java | 7 -------
.../flink/streaming/util/MockStreamingRuntimeContext.java | 7 +++++++
.../streaming/api/functions/async/RichAsyncFunction.java | 7 -------
pom.xml | 4 ++++
7 files changed, 11 insertions(+), 39 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java
index e783d293864..66970e2440d 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java
@@ -20,7 +20,6 @@ package org.apache.flink.api.common.functions;
import org.apache.flink.annotation.Public;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobInfo;
import org.apache.flink.api.common.TaskInfo;
@@ -179,17 +178,6 @@ public interface RuntimeContext {
return getTaskInfo().getTaskNameWithSubtasks();
}
- /**
- * Returns the {@link org.apache.flink.api.common.ExecutionConfig} for the currently executing
- * job.
- *
- * @deprecated This method has been deprecated and will be removed in the upcoming FLINK major
- * version FLINK-2.0. Users relying on this method should migrate to alternative getter
- * methods, such as {@link #getGlobalJobParameters()} or {@link #isObjectReuseEnabled()}.
- */
- @Deprecated
- ExecutionConfig getExecutionConfig();
-
/**
* Create a serializer for a given type.
*
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/util/AbstractRuntimeUDFContext.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/util/AbstractRuntimeUDFContext.java
index a5e36599b89..75efc2a82d8 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/util/AbstractRuntimeUDFContext.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/functions/util/AbstractRuntimeUDFContext.java
@@ -90,12 +90,6 @@ public abstract class AbstractRuntimeUDFContext implements RuntimeContext {
this.metrics = metrics;
}
- @Override
- @Deprecated
- public ExecutionConfig getExecutionConfig() {
- return executionConfig;
- }
-
@Override
public <T> TypeSerializer<T> createSerializer(TypeInformation<T> typeInformation) {
return typeInformation.createSerializer(executionConfig.getSerializerConfig());
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepRuntimeContext.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepRuntimeContext.java
index fb33f9c5eb6..ee8f7dd77dd 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepRuntimeContext.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepRuntimeContext.java
@@ -19,7 +19,6 @@
package org.apache.flink.cep.operator;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobInfo;
import org.apache.flink.api.common.TaskInfo;
import org.apache.flink.api.common.accumulators.Accumulator;
@@ -73,12 +72,6 @@ class CepRuntimeContext implements RuntimeContext {
return runtimeContext.getMetricGroup();
}
- @Override
- @Deprecated
- public ExecutionConfig getExecutionConfig() {
- return runtimeContext.getExecutionConfig();
- }
-
@Override
public <T> TypeSerializer<T> createSerializer(TypeInformation<T> typeInformation) {
return runtimeContext.createSerializer(typeInformation);
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointRuntimeContext.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointRuntimeContext.java
index 8190c9b0c13..e39d18c68a6 100644
--- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointRuntimeContext.java
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointRuntimeContext.java
@@ -19,7 +19,6 @@
package org.apache.flink.state.api.runtime;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobInfo;
import org.apache.flink.api.common.TaskInfo;
import org.apache.flink.api.common.accumulators.Accumulator;
@@ -98,12 +97,6 @@ public final class SavepointRuntimeContext implements RuntimeContext {
return ctx.getMetricGroup();
}
- @Override
- @Deprecated
- public ExecutionConfig getExecutionConfig() {
- return ctx.getExecutionConfig();
- }
-
@Override
public <T> TypeSerializer<T> createSerializer(TypeInformation<T> typeInformation) {
return ctx.createSerializer(typeInformation);
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/util/MockStreamingRuntimeContext.java b/flink-runtime/src/test/java/org/apache/flink/streaming/util/MockStreamingRuntimeContext.java
index 6676a3b4526..887ed968171 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/util/MockStreamingRuntimeContext.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/util/MockStreamingRuntimeContext.java
@@ -39,6 +39,8 @@ public class MockStreamingRuntimeContext extends StreamingRuntimeContext {
private final int numParallelSubtasks;
private final int subtaskIndex;
+ private final MockEnvironment environment;
+
public MockStreamingRuntimeContext(
boolean isCheckpointingEnabled, int numParallelSubtasks, int subtaskIndex) {
@@ -66,6 +68,11 @@ public class MockStreamingRuntimeContext extends StreamingRuntimeContext {
this.isCheckpointingEnabled = isCheckpointingEnabled;
this.numParallelSubtasks = numParallelSubtasks;
this.subtaskIndex = subtaskIndex;
+ this.environment = environment;
+ }
+
+ public ExecutionConfig getExecutionConfig() {
+ return environment.getExecutionConfig();
}
@Override
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/RichAsyncFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/RichAsyncFunction.java
index 2a727c69dcf..5f9b33fe7e5 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/RichAsyncFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/RichAsyncFunction.java
@@ -19,7 +19,6 @@
package org.apache.flink.streaming.api.functions.async;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobInfo;
import org.apache.flink.api.common.TaskInfo;
import org.apache.flink.api.common.accumulators.Accumulator;
@@ -114,12 +113,6 @@ public abstract class RichAsyncFunction<IN, OUT> extends AbstractRichFunction
return runtimeContext.getMetricGroup();
}
- @Override
- @Deprecated
- public ExecutionConfig getExecutionConfig() {
- return runtimeContext.getExecutionConfig();
- }
-
@Override
public <T> TypeSerializer<T> createSerializer(TypeInformation<T> typeInformation) {
return runtimeContext.createSerializer(typeInformation);
diff --git a/pom.xml b/pom.xml
index 3c2aabc8478..777fea41f31 100644
--- a/pom.xml
+++ b/pom.xml
@@ -2376,6 +2376,10 @@ under the License.
<exclude>org.apache.flink.streaming.api.functions.source.RichSourceFunction</exclude>
<exclude>org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction</exclude>
<exclude>org.apache.flink.streaming.api.functions.source.ParallelSourceFunction</exclude>
+ <!-- FLINK-33712 deprecated the RuntimeContext#getExecutionConfig; this will be removed in Flink 2.0. -->
+ <exclude>
+ org.apache.flink.api.common.functions.RuntimeContext#getExecutionConfig()
+ </exclude>
<!-- MARKER: end exclusions -->
</excludes>
<accessModifier>public</accessModifier>