This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.15 by this push:
new 74bc6d2f776 [FLINK-29500] InitializeOnMaster uses wrong parallelism
with AdaptiveScheduler
74bc6d2f776 is described below
commit 74bc6d2f776baad35b2f7f5be41cdfe87759192f
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 | 69 ++++++++++++++++++++--
.../runtime/minicluster/MiniClusterITCase.java | 6 +-
.../example/failing/JobSubmissionFailsITCase.java | 2 +-
10 files changed, 154 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 c17317a70b8..edb74e7e011 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
@@ -1146,7 +1146,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 d54a1243dc7..5dbb0eb5976 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
@@ -169,7 +169,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 843ed765daf..83aa6ce88d5 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
@@ -520,19 +520,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 542d37d57b4..970dfdee71f 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
@@ -1418,8 +1418,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 29594c369e1..f75f557503b 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
@@ -67,8 +67,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());
}
@@ -98,7 +98,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 47cc722fa9b..d0b91c1d33d 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,27 @@ 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.SharedObjects;
+import org.apache.flink.testutils.junit.SharedReference;
import org.apache.flink.util.InstantiationUtil;
+import org.junit.Rule;
import org.junit.Test;
import java.io.IOException;
import java.net.URL;
import java.net.URLClassLoader;
+import java.util.concurrent.atomic.AtomicInteger;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
@@ -44,6 +51,8 @@ import static org.junit.Assert.fail;
@SuppressWarnings("serial")
public class JobTaskVertexTest {
+ @Rule public final SharedObjects sharedObjects = SharedObjects.create();
+
@Test
public void testConnectDirectly() {
JobVertex source = new JobVertex("source");
@@ -80,7 +89,8 @@ public class JobTaskVertexTest {
final ClassLoader cl = new TestClassLoader();
try {
- vertex.initializeOnMaster(cl);
+ vertex.initializeOnMaster(
+ new SimpleInitializeOnMasterContext(cl,
vertex.getParallelism()));
fail("Did not throw expected exception.");
} catch (TestException e) {
// all good
@@ -89,7 +99,8 @@ public class JobTaskVertexTest {
InputOutputFormatVertex copy = InstantiationUtil.clone(vertex);
ClassLoader ctxCl = Thread.currentThread().getContextClassLoader();
try {
- copy.initializeOnMaster(cl);
+ copy.initializeOnMaster(
+ new SimpleInitializeOnMasterContext(cl,
copy.getParallelism()));
fail("Did not throw expected exception.");
} catch (TestException e) {
// all good
@@ -100,7 +111,8 @@ public class JobTaskVertexTest {
Thread.currentThread().getContextClassLoader());
try {
- copy.finalizeOnMaster(cl);
+ copy.finalizeOnMaster(
+ new SimpleInitializeOnMasterContext(cl,
copy.getParallelism()));
fail("Did not throw expected exception.");
} catch (TestException e) {
// all good
@@ -130,7 +142,8 @@ public class JobTaskVertexTest {
final ClassLoader cl = new TestClassLoader();
- vertex.initializeOnMaster(cl);
+ vertex.initializeOnMaster(
+ new SimpleInitializeOnMasterContext(cl,
vertex.getParallelism()));
InputSplit[] splits =
vertex.getInputSplitSource().createInputSplits(77);
assertNotNull(splits);
@@ -142,8 +155,56 @@ public class JobTaskVertexTest {
}
}
+ @Test
+ public 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.");
}
}
