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 e1147ca7e390c413c84d11a15b9f09834e930d06 Author: Weijie Guo <[email protected]> AuthorDate: Fri Mar 22 16:41:11 2024 +0800 [FLINK-34548][API] Introduce ExecutionEnvironment --- flink-clients/pom.xml | 6 + .../java/org/apache/flink/client/ClientUtils.java | 7 + .../flink/api/common/RuntimeExecutionMode.java | 0 .../flink/datastream/api/ExecutionEnvironment.java | 53 +++++ flink-datastream/pom.xml | 29 +++ .../impl/ExecutionContextEnvironment.java | 58 +++++ .../impl/ExecutionEnvironmentFactory.java | 33 +++ .../datastream/impl/ExecutionEnvironmentImpl.java | 244 +++++++++++++++++++++ .../impl/ExecutionEnvironmentImplTest.java | 70 ++++++ .../impl/TestingExecutionEnvironmentFactory.java | 39 ++++ .../datastream/impl/TestingTransformation.java | 43 ++++ 11 files changed, 582 insertions(+) diff --git a/flink-clients/pom.xml b/flink-clients/pom.xml index 03d7891d96f..07711da1eca 100644 --- a/flink-clients/pom.xml +++ b/flink-clients/pom.xml @@ -95,6 +95,12 @@ under the License. <version>${project.version}</version> </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-datastream</artifactId> + <version>${project.version}</version> + </dependency> + <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-core</artifactId> diff --git a/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java b/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java index a104d041eff..88fcc98e3ac 100644 --- a/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java +++ b/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java @@ -31,6 +31,7 @@ import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.DeploymentOptions; import org.apache.flink.core.execution.JobClient; import org.apache.flink.core.execution.PipelineExecutorServiceLoader; +import org.apache.flink.datastream.impl.ExecutionContextEnvironment; import org.apache.flink.runtime.client.JobInitializationException; import org.apache.flink.runtime.jobmaster.JobResult; import org.apache.flink.runtime.rest.HttpHeader; @@ -104,11 +105,17 @@ public enum ClientUtils { enforceSingleJobExecution, suppressSysout); + // For DataStream v2. + ExecutionContextEnvironment.setAsContext( + executorServiceLoader, configuration, userCodeClassLoader); + try { program.invokeInteractiveModeForExecution(); } finally { ContextEnvironment.unsetAsContext(); StreamContextEnvironment.unsetAsContext(); + // For DataStream v2. + ExecutionContextEnvironment.unsetAsContext(); } } finally { Thread.currentThread().setContextClassLoader(contextClassLoader); diff --git a/flink-core/src/main/java/org/apache/flink/api/common/RuntimeExecutionMode.java b/flink-core-api/src/main/java/org/apache/flink/api/common/RuntimeExecutionMode.java similarity index 100% rename from flink-core/src/main/java/org/apache/flink/api/common/RuntimeExecutionMode.java rename to flink-core-api/src/main/java/org/apache/flink/api/common/RuntimeExecutionMode.java diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/ExecutionEnvironment.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/ExecutionEnvironment.java new file mode 100644 index 00000000000..58515be96fb --- /dev/null +++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/ExecutionEnvironment.java @@ -0,0 +1,53 @@ +/* + * 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.datastream.api; + +import org.apache.flink.annotation.Experimental; +import org.apache.flink.api.common.RuntimeExecutionMode; + +/** + * This is the context in which a program is executed. + * + * <p>The environment provides methods to create a DataStream and control the job execution. + */ +@Experimental +public interface ExecutionEnvironment { + /** + * Get the execution environment instance. + * + * @return A {@link ExecutionEnvironment} instance. + */ + static ExecutionEnvironment getInstance() throws ReflectiveOperationException { + return (ExecutionEnvironment) + Class.forName("org.apache.flink.datastream.impl.ExecutionEnvironmentImpl") + .getMethod("newInstance") + .invoke(null); + } + + /** Execute and submit the job attached to this environment. */ + void execute(String jobName) throws Exception; + + /** Get the execution mode of this environment. */ + RuntimeExecutionMode getExecutionMode(); + + /** Set the execution mode for this environment. */ + ExecutionEnvironment setExecutionMode(RuntimeExecutionMode runtimeMode); + + // TODO introduce method to add source +} diff --git a/flink-datastream/pom.xml b/flink-datastream/pom.xml index 5fc9b908638..80a13c8317a 100644 --- a/flink-datastream/pom.xml +++ b/flink-datastream/pom.xml @@ -40,5 +40,34 @@ under the License. <artifactId>flink-datastream-api</artifactId> <version>${project.version}</version> </dependency> + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-streaming-java</artifactId> + <version>${project.version}</version> + </dependency> + + <!-- Test Dependencies --> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-test-utils-junit</artifactId> + <scope>test</scope> + </dependency> + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-streaming-java</artifactId> + <version>${project.version}</version> + <type>test-jar</type> + <scope>test</scope> + </dependency> + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-runtime</artifactId> + <version>${project.version}</version> + <type>test-jar</type> + <scope>test</scope> + </dependency> </dependencies> </project> diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/ExecutionContextEnvironment.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/ExecutionContextEnvironment.java new file mode 100644 index 00000000000..efd0b92a024 --- /dev/null +++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/ExecutionContextEnvironment.java @@ -0,0 +1,58 @@ +/* + * 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.datastream.impl; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.core.execution.PipelineExecutorServiceLoader; +import org.apache.flink.datastream.api.ExecutionEnvironment; + +/** + * Special {@link ExecutionEnvironment} that will be used in cases where the CLI client or testing + * utilities create a {@link ExecutionEnvironment} that should be used when {@link + * ExecutionEnvironment#getInstance()} ()} is called. + */ +public class ExecutionContextEnvironment extends ExecutionEnvironmentImpl { + public ExecutionContextEnvironment( + final PipelineExecutorServiceLoader executorServiceLoader, + final Configuration configuration, + final ClassLoader userCodeClassLoader) { + super(executorServiceLoader, configuration, userCodeClassLoader); + } + + // -------------------------------------------------------------------------------------------- + + public static void setAsContext( + final PipelineExecutorServiceLoader executorServiceLoader, + final Configuration clusterConfiguration, + final ClassLoader userCodeClassLoader) { + final ExecutionEnvironmentFactory factory = + envInitConfig -> { + final Configuration mergedEnvConfig = new Configuration(); + mergedEnvConfig.addAll(clusterConfiguration); + mergedEnvConfig.addAll(envInitConfig); + return new ExecutionContextEnvironment( + executorServiceLoader, mergedEnvConfig, userCodeClassLoader); + }; + initializeContextEnvironment(factory); + } + + public static void unsetAsContext() { + resetContextEnvironment(); + } +} diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/ExecutionEnvironmentFactory.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/ExecutionEnvironmentFactory.java new file mode 100644 index 00000000000..c2877deec7c --- /dev/null +++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/ExecutionEnvironmentFactory.java @@ -0,0 +1,33 @@ +/* + * 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.datastream.impl; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.datastream.api.ExecutionEnvironment; + +/** Factory class for execution environments. */ +@FunctionalInterface +public interface ExecutionEnvironmentFactory { + /** + * Creates a ExecutionEnvironment from this factory. + * + * @return A ExecutionEnvironment. + */ + ExecutionEnvironment createExecutionEnvironment(Configuration configuration); +} diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/ExecutionEnvironmentImpl.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/ExecutionEnvironmentImpl.java new file mode 100644 index 00000000000..706837c55e8 --- /dev/null +++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/ExecutionEnvironmentImpl.java @@ -0,0 +1,244 @@ +/* + * 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.datastream.impl; + +import org.apache.flink.api.common.ExecutionConfig; +import org.apache.flink.api.common.RuntimeExecutionMode; +import org.apache.flink.api.dag.Transformation; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.DeploymentOptions; +import org.apache.flink.configuration.ExecutionOptions; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.core.execution.DefaultExecutorServiceLoader; +import org.apache.flink.core.execution.JobClient; +import org.apache.flink.core.execution.PipelineExecutor; +import org.apache.flink.core.execution.PipelineExecutorFactory; +import org.apache.flink.core.execution.PipelineExecutorServiceLoader; +import org.apache.flink.datastream.api.ExecutionEnvironment; +import org.apache.flink.streaming.api.environment.CheckpointConfig; +import org.apache.flink.streaming.api.graph.StreamGraph; +import org.apache.flink.streaming.api.graph.StreamGraphGenerator; +import org.apache.flink.util.ExceptionUtils; +import org.apache.flink.util.FlinkException; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; + +import static org.apache.flink.streaming.api.graph.StreamGraphGenerator.DEFAULT_TIME_CHARACTERISTIC; +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * The implementation of {@link ExecutionEnvironment}. + * + * <p>IMPORTANT: Even though this is not part of public API, {@link ExecutionEnvironment} will get + * this class instance through reflection, so we must ensure that the package path, class name and + * the signature of {@link #newInstance()} does not change. + */ +public class ExecutionEnvironmentImpl implements ExecutionEnvironment { + private final List<Transformation<?>> transformations = new ArrayList<>(); + + private final ExecutionConfig executionConfig; + + /** Settings that control the checkpointing behavior. */ + private final CheckpointConfig checkpointCfg; + + private final Configuration configuration; + + private final ClassLoader userClassloader; + + private final PipelineExecutorServiceLoader executorServiceLoader; + + /** + * The environment of the context (local by default, cluster if invoked through command line). + */ + private static ExecutionEnvironmentFactory contextEnvironmentFactory = null; + + /** + * Create and return an instance of {@link ExecutionEnvironment}. + * + * <p>IMPORTANT: The method is only expected to be called by {@link ExecutionEnvironment} via + * reflection, so we must ensure that the package path, class name and the signature of this + * method does not change. + */ + public static ExecutionEnvironment newInstance() { + if (contextEnvironmentFactory != null) { + return contextEnvironmentFactory.createExecutionEnvironment(new Configuration()); + } else { + final Configuration configuration = new Configuration(); + configuration.set(DeploymentOptions.TARGET, "local"); + configuration.set(DeploymentOptions.ATTACHED, true); + return new ExecutionEnvironmentImpl( + new DefaultExecutorServiceLoader(), configuration, null); + } + } + + ExecutionEnvironmentImpl( + PipelineExecutorServiceLoader executorServiceLoader, + Configuration configuration, + ClassLoader classLoader) { + this.executorServiceLoader = checkNotNull(executorServiceLoader); + this.configuration = configuration; + this.executionConfig = new ExecutionConfig(this.configuration); + this.checkpointCfg = new CheckpointConfig(this.configuration); + this.userClassloader = classLoader == null ? getClass().getClassLoader() : classLoader; + configure(configuration, userClassloader); + } + + @Override + public void execute(String jobName) throws Exception { + StreamGraph streamGraph = getStreamGraph(); + if (jobName != null) { + streamGraph.setJobName(jobName); + } + + execute(streamGraph); + } + + @Override + public RuntimeExecutionMode getExecutionMode() { + return configuration.get(ExecutionOptions.RUNTIME_MODE); + } + + @Override + public ExecutionEnvironment setExecutionMode(RuntimeExecutionMode runtimeMode) { + checkNotNull(runtimeMode); + configuration.set(ExecutionOptions.RUNTIME_MODE, runtimeMode); + return this; + } + + protected static void initializeContextEnvironment(ExecutionEnvironmentFactory ctx) { + contextEnvironmentFactory = ctx; + } + + protected static void resetContextEnvironment() { + contextEnvironmentFactory = null; + } + + public Configuration getConfiguration() { + return this.configuration; + } + + public ExecutionConfig getExecutionConfig() { + return executionConfig; + } + + public int getParallelism() { + return executionConfig.getParallelism(); + } + + public List<Transformation<?>> getTransformations() { + return transformations; + } + + public void setParallelism(int parallelism) { + executionConfig.setParallelism(parallelism); + } + + // ----------------------------------------------- + // Internal Methods + // ----------------------------------------------- + + public void addOperator(Transformation<?> transformation) { + checkNotNull(transformation, "transformation must not be null."); + this.transformations.add(transformation); + } + + private void execute(StreamGraph streamGraph) throws Exception { + final JobClient jobClient = executeAsync(streamGraph); + + try { + if (configuration.get(DeploymentOptions.ATTACHED)) { + jobClient.getJobExecutionResult().get(); + } + // TODO Supports accumulator. + } catch (Throwable t) { + // get() on the JobExecutionResult Future will throw an ExecutionException. This + // behaviour was largely not there in Flink versions before the PipelineExecutor + // refactoring so we should strip that exception. + Throwable strippedException = ExceptionUtils.stripExecutionException(t); + ExceptionUtils.rethrowException(strippedException); + } + } + + private JobClient executeAsync(StreamGraph streamGraph) throws Exception { + checkNotNull(streamGraph, "StreamGraph cannot be null."); + final PipelineExecutor executor = getPipelineExecutor(); + + CompletableFuture<JobClient> jobClientFuture = + executor.execute(streamGraph, configuration, getClass().getClassLoader()); + + try { + // TODO Supports job listeners. + return jobClientFuture.get(); + } catch (ExecutionException executionException) { + final Throwable strippedException = + ExceptionUtils.stripExecutionException(executionException); + throw new FlinkException( + String.format("Failed to execute job '%s'.", streamGraph.getJobName()), + strippedException); + } + } + + /** Get {@link StreamGraph} and clear all transformations. */ + public StreamGraph getStreamGraph() { + final StreamGraph streamGraph = getStreamGraphGenerator(transformations).generate(); + transformations.clear(); + return streamGraph; + } + + private StreamGraphGenerator getStreamGraphGenerator(List<Transformation<?>> transformations) { + if (transformations.size() <= 0) { + throw new IllegalStateException( + "No operators defined in streaming topology. Cannot execute."); + } + + // We copy the transformation so that newly added transformations cannot intervene with the + // stream graph generation. + return new StreamGraphGenerator( + new ArrayList<>(transformations), + executionConfig, + checkpointCfg, + configuration) + .setTimeCharacteristic(DEFAULT_TIME_CHARACTERISTIC); + } + + private PipelineExecutor getPipelineExecutor() throws Exception { + checkNotNull( + configuration.get(DeploymentOptions.TARGET), + "No execution.target specified in your configuration file."); + + final PipelineExecutorFactory executorFactory = + executorServiceLoader.getExecutorFactory(configuration); + + checkNotNull( + executorFactory, + "Cannot find compatible factory for specified execution.target (=%s)", + configuration.get(DeploymentOptions.TARGET)); + + return executorFactory.getExecutor(configuration); + } + + private void configure(ReadableConfig configuration, ClassLoader classLoader) { + this.configuration.addAll(Configuration.fromMap(configuration.toMap())); + executionConfig.configure(configuration, classLoader); + checkpointCfg.configure(configuration); + } +} diff --git a/flink-datastream/src/test/java/org/apache/flink/datastream/impl/ExecutionEnvironmentImplTest.java b/flink-datastream/src/test/java/org/apache/flink/datastream/impl/ExecutionEnvironmentImplTest.java new file mode 100644 index 00000000000..65e57ed1d9a --- /dev/null +++ b/flink-datastream/src/test/java/org/apache/flink/datastream/impl/ExecutionEnvironmentImplTest.java @@ -0,0 +1,70 @@ +/* + * 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.datastream.impl; + +import org.apache.flink.api.common.RuntimeExecutionMode; +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.core.execution.DefaultExecutorServiceLoader; +import org.apache.flink.datastream.api.ExecutionEnvironment; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link ExecutionEnvironmentImpl}. */ +class ExecutionEnvironmentImplTest { + @Test + void testSetContextExecutionEnvironment() throws Exception { + ExecutionEnvironment expectedEnv = + new ExecutionEnvironmentImpl( + new DefaultExecutorServiceLoader(), new Configuration(), null); + TestingExecutionEnvironmentFactory factory = + new TestingExecutionEnvironmentFactory((config) -> expectedEnv); + ExecutionEnvironmentImpl.initializeContextEnvironment(factory); + ExecutionEnvironment env = ExecutionEnvironment.getInstance(); + assertThat(env).isSameAs(expectedEnv); + + ExecutionEnvironmentImpl.resetContextEnvironment(); + ExecutionEnvironment env2 = ExecutionEnvironment.getInstance(); + assertThat(env2).isNotSameAs(expectedEnv); + } + + @Test + void testAddOperator() { + ExecutionEnvironmentImpl env = + new ExecutionEnvironmentImpl( + new DefaultExecutorServiceLoader(), new Configuration(), null); + + TestingTransformation<Integer> t1 = new TestingTransformation<>("t1", Types.INT, 10); + TestingTransformation<String> t2 = new TestingTransformation<>("t2", Types.STRING, 5); + env.addOperator(t1); + env.addOperator(t2); + assertThat(env.getTransformations()).containsExactly(t1, t2); + } + + @Test + void testSetExecutionMode() { + ExecutionEnvironmentImpl env = + new ExecutionEnvironmentImpl( + new DefaultExecutorServiceLoader(), new Configuration(), null); + env.setExecutionMode(RuntimeExecutionMode.BATCH); + assertThat(env.getExecutionMode()).isEqualTo(RuntimeExecutionMode.BATCH); + } +} diff --git a/flink-datastream/src/test/java/org/apache/flink/datastream/impl/TestingExecutionEnvironmentFactory.java b/flink-datastream/src/test/java/org/apache/flink/datastream/impl/TestingExecutionEnvironmentFactory.java new file mode 100644 index 00000000000..c799f3e6986 --- /dev/null +++ b/flink-datastream/src/test/java/org/apache/flink/datastream/impl/TestingExecutionEnvironmentFactory.java @@ -0,0 +1,39 @@ +/* + * 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.datastream.impl; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.datastream.api.ExecutionEnvironment; + +import java.util.function.Function; + +/** Mock implementation of {@link ExecutionEnvironmentFactory} for testing. */ +public class TestingExecutionEnvironmentFactory implements ExecutionEnvironmentFactory { + private final Function<Configuration, ExecutionEnvironment> createExecutionEnvironmentFunction; + + public TestingExecutionEnvironmentFactory( + Function<Configuration, ExecutionEnvironment> createExecutionEnvironmentFunction) { + this.createExecutionEnvironmentFunction = createExecutionEnvironmentFunction; + } + + @Override + public ExecutionEnvironment createExecutionEnvironment(Configuration configuration) { + return createExecutionEnvironmentFunction.apply(configuration); + } +} diff --git a/flink-datastream/src/test/java/org/apache/flink/datastream/impl/TestingTransformation.java b/flink-datastream/src/test/java/org/apache/flink/datastream/impl/TestingTransformation.java new file mode 100644 index 00000000000..20b596a8631 --- /dev/null +++ b/flink-datastream/src/test/java/org/apache/flink/datastream/impl/TestingTransformation.java @@ -0,0 +1,43 @@ +/* + * 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.datastream.impl; + +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.dag.Transformation; + +import java.util.Collections; +import java.util.List; + +/** Mock {@link Transformation} for testing. */ +public class TestingTransformation<T> extends Transformation<T> { + + public TestingTransformation(String name, TypeInformation<T> outputType, int parallelism) { + super(name, outputType, parallelism); + } + + @Override + public List<Transformation<?>> getTransitivePredecessors() { + return Collections.emptyList(); + } + + @Override + public List<Transformation<?>> getInputs() { + return Collections.emptyList(); + } +}
