This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new cadfab59a9a [FLINK-29500] InitializeOnMaster uses wrong parallelism
with AdaptiveScheduler
cadfab59a9a is described below
commit cadfab59a9a5f6eaaaae7f95e2a9fb3ad7b1b1d9
Author: Dawid Wysakowicz <[email protected]>
AuthorDate: Tue Oct 4 13:50:28 2022 +0200
[FLINK-29500] InitializeOnMaster uses wrong parallelism with
AdaptiveScheduler
We change the signature of JobVertex#initializeOnMaster/finalizeOnMaster to
pass a Context object. In this context we can pass
the actual parallelism the vertex will be run with.
---
.../executiongraph/DefaultExecutionGraph.java | 5 +-
.../DefaultExecutionGraphBuilder.java | 7 +-
.../SimpleInitializeOnMasterContext.java | 45 ++++++++++++
.../runtime/jobgraph/InputOutputFormatVertex.java | 12 ++--
.../apache/flink/runtime/jobgraph/JobVertex.java | 21 ++++--
.../flink/runtime/dispatcher/DispatcherTest.java | 4 +-
.../executiongraph/FinalizeOnMasterTest.java | 6 +-
.../flink/runtime/jobgraph/JobTaskVertexTest.java | 79 ++++++++++++++++++++--
.../runtime/minicluster/MiniClusterITCase.java | 6 +-
.../example/failing/JobSubmissionFailsITCase.java | 2 +-
10 files changed, 164 insertions(+), 23 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
index 164ff1eb567..3e4efea3758 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
@@ -1215,7 +1215,10 @@ public class DefaultExecutionGraph implements
ExecutionGraph, InternalExecutionG
try {
for (ExecutionJobVertex ejv : verticesInCreationOrder) {
- ejv.getJobVertex().finalizeOnMaster(getUserClassLoader());
+ ejv.getJobVertex()
+ .finalizeOnMaster(
+ new SimpleInitializeOnMasterContext(
+ getUserClassLoader(),
ejv.getParallelism()));
}
} catch (Throwable t) {
ExceptionUtils.rethrowIfFatalError(t);
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphBuilder.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphBuilder.java
index 462664b2865..9dffbdf2619 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphBuilder.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphBuilder.java
@@ -173,7 +173,12 @@ public class DefaultExecutionGraphBuilder {
}
try {
- vertex.initializeOnMaster(classLoader);
+ vertex.initializeOnMaster(
+ new SimpleInitializeOnMasterContext(
+ classLoader,
+ vertexParallelismStore
+ .getParallelismInfo(vertex.getID())
+ .getParallelism()));
} catch (Throwable t) {
throw new JobExecutionException(
jobId,
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/SimpleInitializeOnMasterContext.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/SimpleInitializeOnMasterContext.java
new file mode 100644
index 00000000000..686300f7aca
--- /dev/null
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/SimpleInitializeOnMasterContext.java
@@ -0,0 +1,45 @@
+/*
+ * 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.executiongraph;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+
+/** A simple pojo based implementation of {@link
JobVertex.InitializeOnMasterContext}. */
+@Internal
+public final class SimpleInitializeOnMasterContext implements
JobVertex.InitializeOnMasterContext {
+
+ private final ClassLoader loader;
+ private final int executionParallelism;
+
+ public SimpleInitializeOnMasterContext(ClassLoader loader, int
executionParallelism) {
+ this.loader = loader;
+ this.executionParallelism = executionParallelism;
+ }
+
+ @Override
+ public ClassLoader getClassLoader() {
+ return loader;
+ }
+
+ @Override
+ public int getExecutionParallelism() {
+ return executionParallelism;
+ }
+}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/InputOutputFormatVertex.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/InputOutputFormatVertex.java
index 097ff181b45..9eb3e36cef7 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/InputOutputFormatVertex.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/InputOutputFormatVertex.java
@@ -53,7 +53,8 @@ public class InputOutputFormatVertex extends JobVertex {
}
@Override
- public void initializeOnMaster(ClassLoader loader) throws Exception {
+ public void initializeOnMaster(InitializeOnMasterContext context) throws
Exception {
+ ClassLoader loader = context.getClassLoader();
final InputOutputFormatContainer formatContainer =
initInputOutputformatContainer(loader);
final ClassLoader original =
Thread.currentThread().getContextClassLoader();
@@ -107,7 +108,8 @@ public class InputOutputFormatVertex extends JobVertex {
}
if (outputFormat instanceof InitializeOnMaster) {
- ((InitializeOnMaster)
outputFormat).initializeGlobal(getParallelism());
+ int executionParallelism =
context.getExecutionParallelism();
+ ((InitializeOnMaster)
outputFormat).initializeGlobal(executionParallelism);
}
}
} finally {
@@ -117,7 +119,8 @@ public class InputOutputFormatVertex extends JobVertex {
}
@Override
- public void finalizeOnMaster(ClassLoader loader) throws Exception {
+ public void finalizeOnMaster(InitializeOnMasterContext context) throws
Exception {
+ final ClassLoader loader = context.getClassLoader();
final InputOutputFormatContainer formatContainer =
initInputOutputformatContainer(loader);
final ClassLoader original =
Thread.currentThread().getContextClassLoader();
@@ -145,7 +148,8 @@ public class InputOutputFormatVertex extends JobVertex {
}
if (outputFormat instanceof FinalizeOnMaster) {
- ((FinalizeOnMaster)
outputFormat).finalizeGlobal(getParallelism());
+ int executionParallelism =
context.getExecutionParallelism();
+ ((FinalizeOnMaster)
outputFormat).finalizeGlobal(executionParallelism);
}
}
} finally {
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertex.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertex.java
index 717bd6d4376..3c7b42fe458 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertex.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertex.java
@@ -559,19 +559,32 @@ public class JobVertex implements java.io.Serializable {
* A hook that can be overwritten by sub classes to implement logic that
is called by the master
* when the job starts.
*
- * @param loader The class loader for user defined code.
+ * @param context Provides contextual information for the initialization
* @throws Exception The method may throw exceptions which cause the job
to fail immediately.
*/
- public void initializeOnMaster(ClassLoader loader) throws Exception {}
+ public void initializeOnMaster(InitializeOnMasterContext context) throws
Exception {}
/**
* A hook that can be overwritten by sub classes to implement logic that
is called by the master
* after the job completed.
*
- * @param loader The class loader for user defined code.
+ * @param context Provides contextual information for the initialization
* @throws Exception The method may throw exceptions which cause the job
to fail immediately.
*/
- public void finalizeOnMaster(ClassLoader loader) throws Exception {}
+ public void finalizeOnMaster(InitializeOnMasterContext context) throws
Exception {}
+
+ public interface InitializeOnMasterContext {
+ /** The class loader for user defined code. */
+ ClassLoader getClassLoader();
+
+ /**
+ * The actual parallelism this vertex will be run with. In contrast,
the {@link
+ * #getParallelism()} is the original parallelism set when creating
the {@link JobGraph} and
+ * might be updated e.g. by the {@link
+ * org.apache.flink.runtime.scheduler.adaptive.AdaptiveScheduler}.
+ */
+ int getExecutionParallelism();
+ }
//
--------------------------------------------------------------------------------------------
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
index a51d8bfc8d8..0f8f130d5fb 100755
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
@@ -1419,8 +1419,8 @@ public class DispatcherTest extends
AbstractDispatcherTest {
}
@Override
- public void initializeOnMaster(ClassLoader loader) throws Exception {
- super.initializeOnMaster(loader);
+ public void initializeOnMaster(InitializeOnMasterContext context)
throws Exception {
+ super.initializeOnMaster(context);
oneShotLatch.await();
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/FinalizeOnMasterTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/FinalizeOnMasterTest.java
index 3524cc14817..a302ce2e862 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/FinalizeOnMasterTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/FinalizeOnMasterTest.java
@@ -77,8 +77,8 @@ public class FinalizeOnMasterTest extends TestLogger {
ExecutionGraphTestUtils.finishAllVertices(eg);
assertEquals(JobStatus.FINISHED, eg.waitUntilTerminal());
- verify(vertex1, times(1)).finalizeOnMaster(any(ClassLoader.class));
- verify(vertex2, times(1)).finalizeOnMaster(any(ClassLoader.class));
+ verify(vertex1,
times(1)).finalizeOnMaster(any(JobVertex.InitializeOnMasterContext.class));
+ verify(vertex2,
times(1)).finalizeOnMaster(any(JobVertex.InitializeOnMasterContext.class));
assertEquals(0, eg.getRegisteredExecutions().size());
}
@@ -109,7 +109,7 @@ public class FinalizeOnMasterTest extends TestLogger {
assertEquals(JobStatus.FAILED, eg.waitUntilTerminal());
- verify(vertex, times(0)).finalizeOnMaster(any(ClassLoader.class));
+ verify(vertex,
times(0)).finalizeOnMaster(any(JobVertex.InitializeOnMasterContext.class));
assertEquals(0, eg.getRegisteredExecutions().size());
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobTaskVertexTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobTaskVertexTest.java
index cf59753999f..d31fdb9d8bf 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobTaskVertexTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobTaskVertexTest.java
@@ -21,20 +21,26 @@ package org.apache.flink.runtime.jobgraph;
import org.apache.flink.api.common.io.FinalizeOnMaster;
import org.apache.flink.api.common.io.GenericInputFormat;
import org.apache.flink.api.common.io.InitializeOnMaster;
+import org.apache.flink.api.common.io.OutputFormat;
import org.apache.flink.api.java.io.DiscardingOutputFormat;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.io.GenericInputSplit;
import org.apache.flink.core.io.InputSplit;
+import org.apache.flink.runtime.executiongraph.SimpleInitializeOnMasterContext;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
import org.apache.flink.runtime.operators.util.TaskConfig;
+import org.apache.flink.testutils.junit.SharedObjectsExtension;
+import org.apache.flink.testutils.junit.SharedReference;
import org.apache.flink.util.InstantiationUtil;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.io.IOException;
import java.net.URL;
import java.net.URLClassLoader;
import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -42,6 +48,8 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
@SuppressWarnings("serial")
class JobTaskVertexTest {
+ @RegisterExtension final SharedObjectsExtension sharedObjects =
SharedObjectsExtension.create();
+
@Test
void testMultipleConsumersVertices() {
JobVertex producer = new JobVertex("producer");
@@ -117,17 +125,32 @@ class JobTaskVertexTest {
final ClassLoader cl = new TestClassLoader();
- assertThatThrownBy(() ->
vertex.initializeOnMaster(cl)).isInstanceOf(TestException.class);
+ assertThatThrownBy(
+ () ->
+ vertex.initializeOnMaster(
+ new SimpleInitializeOnMasterContext(
+ cl, vertex.getParallelism())))
+ .isInstanceOf(TestException.class);
InputOutputFormatVertex copy = InstantiationUtil.clone(vertex);
ClassLoader ctxCl = Thread.currentThread().getContextClassLoader();
- assertThatThrownBy(() ->
copy.initializeOnMaster(cl)).isInstanceOf(TestException.class);
+ assertThatThrownBy(
+ () ->
+ copy.initializeOnMaster(
+ new SimpleInitializeOnMasterContext(
+ cl, copy.getParallelism())))
+ .isInstanceOf(TestException.class);
assertThat(Thread.currentThread().getContextClassLoader())
.as("Previous classloader was not restored.")
.isEqualTo(ctxCl);
- assertThatThrownBy(() ->
copy.finalizeOnMaster(cl)).isInstanceOf(TestException.class);
+ assertThatThrownBy(
+ () ->
+ copy.finalizeOnMaster(
+ new SimpleInitializeOnMasterContext(
+ cl, copy.getParallelism())))
+ .isInstanceOf(TestException.class);
assertThat(Thread.currentThread().getContextClassLoader())
.as("Previous classloader was not restored.")
.isEqualTo(ctxCl);
@@ -147,7 +170,7 @@ class JobTaskVertexTest {
final ClassLoader cl = new TestClassLoader();
- vertex.initializeOnMaster(cl);
+ vertex.initializeOnMaster(new SimpleInitializeOnMasterContext(cl,
vertex.getParallelism()));
InputSplit[] splits =
vertex.getInputSplitSource().createInputSplits(77);
assertThat(splits).isNotNull();
@@ -155,8 +178,56 @@ class JobTaskVertexTest {
assertThat(splits[0].getClass()).isEqualTo(TestSplit.class);
}
+ @Test
+ void testOutputFormatUsesCorrectParallelism() throws Exception {
+ final InputOutputFormatVertex vertex = new
InputOutputFormatVertex("Name");
+ int initialParallelism = 1;
+ vertex.setParallelism(initialParallelism);
+
+ OperatorID operatorID = new OperatorID();
+ // just a mutable container for integer
+ SharedReference<AtomicInteger> globalParallelism =
sharedObjects.add(new AtomicInteger());
+ new
InputOutputFormatContainer(Thread.currentThread().getContextClassLoader())
+ .addOutputFormat(operatorID, new
TestInitializeOutputFormat(globalParallelism))
+ .write(new TaskConfig(vertex.getConfiguration()));
+
+ int executionParallelism = initialParallelism + 3;
+ try (final TestClassLoader cl = new TestClassLoader()) {
+ vertex.initializeOnMaster(
+ new SimpleInitializeOnMasterContext(cl,
executionParallelism));
+
assertThat(globalParallelism.get().get()).isEqualTo(executionParallelism);
+ }
+ }
+
//
--------------------------------------------------------------------------------------------
+ private static final class TestInitializeOutputFormat
+ implements OutputFormat<Object>, InitializeOnMaster {
+
+ private final SharedReference<AtomicInteger> globalParallelism;
+
+ private TestInitializeOutputFormat(SharedReference<AtomicInteger>
globalParallelism) {
+ this.globalParallelism = globalParallelism;
+ }
+
+ @Override
+ public void configure(Configuration parameters) {}
+
+ @Override
+ public void open(int taskNumber, int numTasks) throws IOException {}
+
+ @Override
+ public void writeRecord(Object record) throws IOException {}
+
+ @Override
+ public void close() throws IOException {}
+
+ @Override
+ public void initializeGlobal(int parallelism) throws IOException {
+ globalParallelism.get().set(parallelism);
+ }
+ }
+
private static final class TestException extends IOException {}
private static final class TestSplit extends GenericInputSplit {
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/MiniClusterITCase.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/MiniClusterITCase.java
index 4efc67785f8..ccf600fea09 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/MiniClusterITCase.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/MiniClusterITCase.java
@@ -714,7 +714,7 @@ public class MiniClusterITCase extends TestLogger {
}
@Override
- public void finalizeOnMaster(ClassLoader loader) throws Exception {
+ public void finalizeOnMaster(InitializeOnMasterContext context) throws
Exception {
Thread.sleep(waitingTime);
finalizedOnMaster.set(true);
}
@@ -731,7 +731,7 @@ public class MiniClusterITCase extends TestLogger {
}
@Override
- public void finalizeOnMaster(ClassLoader loader) {
+ public void finalizeOnMaster(InitializeOnMasterContext context) {
throw new OutOfMemoryError("Java heap space");
}
}
@@ -743,7 +743,7 @@ public class MiniClusterITCase extends TestLogger {
}
@Override
- public void initializeOnMaster(ClassLoader loader) {
+ public void initializeOnMaster(InitializeOnMasterContext context) {
throw new OutOfMemoryError("Java heap space");
}
}
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/example/failing/JobSubmissionFailsITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/example/failing/JobSubmissionFailsITCase.java
index eb41e95a9b7..402ae192e60 100644
---
a/flink-tests/src/test/java/org/apache/flink/test/example/failing/JobSubmissionFailsITCase.java
+++
b/flink-tests/src/test/java/org/apache/flink/test/example/failing/JobSubmissionFailsITCase.java
@@ -147,7 +147,7 @@ public class JobSubmissionFailsITCase extends TestLogger {
}
@Override
- public void initializeOnMaster(ClassLoader loader) throws Exception {
+ public void initializeOnMaster(InitializeOnMasterContext context)
throws Exception {
throw new Exception("Test exception.");
}
}
