You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by wa...@apache.org on 2023/02/16 15:15:06 UTC
[flink] branch release-1.17 updated (9e1cf08e246 -> 629bc9a3d35)
This is an automated email from the ASF dual-hosted git repository.
wanglijie pushed a change to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
from 9e1cf08e246 [FLINK-30895][coordination] Dynamically adjust slot distribution
new 35876807595 [hotfix] Fix the missing parameter error in ForwardForConsecutiveHashPartitionerTest
new fceeec76066 [hotfix] Refactor code to configure stream graph at the same location
new 629bc9a3d35 [FLINK-31055][runtime] Fix the bug that dynamic flag of stream graph does not take effect when translating the transformations
The 3 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:
.../streaming/api/graph/StreamGraphGenerator.java | 31 ++++++------
.../ForwardForConsecutiveHashPartitionerTest.java | 9 ++--
.../ForwardForUnspecifiedPartitionerTest.java | 55 +++++++++++++++++++---
3 files changed, 70 insertions(+), 25 deletions(-)
[flink] 02/03: [hotfix] Refactor code to configure stream graph at the same location
Posted by wa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wanglijie pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
commit fceeec76066114f1acfbb9ec96c76c21dd9787d0
Author: Lijie Wang <wa...@gmail.com>
AuthorDate: Thu Feb 16 12:01:02 2023 +0800
[hotfix] Refactor code to configure stream graph at the same location
---
.../apache/flink/streaming/api/graph/StreamGraphGenerator.java | 10 +++++-----
1 file changed, 5 insertions(+), 5 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
index ba05b6b9c22..c4f2c579493 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
@@ -310,11 +310,6 @@ public class StreamGraphGenerator {
public StreamGraph generate() {
streamGraph = new StreamGraph(executionConfig, checkpointConfig, savepointRestoreSettings);
- streamGraph.setAutoParallelismEnabled(
- configuration.get(BatchExecutionOptions.ADAPTIVE_AUTO_PARALLELISM_ENABLED));
- streamGraph.setEnableCheckpointsAfterTasksFinish(
- configuration.get(
- ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH));
shouldExecuteInBatchMode = shouldExecuteInBatchMode();
configureStreamGraph(streamGraph);
@@ -368,6 +363,11 @@ public class StreamGraphGenerator {
graph.setVertexDescriptionMode(configuration.get(PipelineOptions.VERTEX_DESCRIPTION_MODE));
graph.setVertexNameIncludeIndexPrefix(
configuration.get(PipelineOptions.VERTEX_NAME_INCLUDE_INDEX_PREFIX));
+ graph.setAutoParallelismEnabled(
+ configuration.get(BatchExecutionOptions.ADAPTIVE_AUTO_PARALLELISM_ENABLED));
+ graph.setEnableCheckpointsAfterTasksFinish(
+ configuration.get(
+ ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH));
if (shouldExecuteInBatchMode) {
configureStreamGraphBatch(graph);
[flink] 01/03: [hotfix] Fix the missing parameter error in ForwardForConsecutiveHashPartitionerTest
Posted by wa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wanglijie pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 35876807595a3bbc5711863bd746c5190b1e16ed
Author: Lijie Wang <wa...@gmail.com>
AuthorDate: Tue Feb 14 21:32:51 2023 +0800
[hotfix] Fix the missing parameter error in ForwardForConsecutiveHashPartitionerTest
---
.../partitioner/ForwardForConsecutiveHashPartitionerTest.java | 9 +++++----
1 file changed, 5 insertions(+), 4 deletions(-)
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/ForwardForConsecutiveHashPartitionerTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/ForwardForConsecutiveHashPartitionerTest.java
index c33adbd92a8..57e7b6238e5 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/ForwardForConsecutiveHashPartitionerTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/ForwardForConsecutiveHashPartitionerTest.java
@@ -23,7 +23,6 @@ import org.apache.flink.streaming.api.graph.NonChainedOutput;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.graph.StreamEdge;
import org.apache.flink.streaming.api.transformations.StreamExchangeMode;
-import org.apache.flink.util.TestLogger;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -31,7 +30,7 @@ import org.junit.jupiter.api.Test;
import java.util.List;
/** Test for {@link ForwardForConsecutiveHashPartitioner}. */
-class ForwardForConsecutiveHashPartitionerTest extends TestLogger {
+class ForwardForConsecutiveHashPartitionerTest {
@Test
void testConvertToForwardPartitioner() {
@@ -46,7 +45,8 @@ class ForwardForConsecutiveHashPartitionerTest extends TestLogger {
"group1",
"group1",
new ForwardForConsecutiveHashPartitioner<>(
- new KeyGroupStreamPartitioner<>(record -> 0L, 100)));
+ new KeyGroupStreamPartitioner<>(record -> 0L, 100)),
+ streamExchangeMode);
List<JobVertex> jobVertices = jobGraph.getVerticesSortedTopologicallyFromSources();
Assertions.assertThat(jobVertices.size()).isEqualTo(1);
JobVertex vertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(0);
@@ -69,7 +69,8 @@ class ForwardForConsecutiveHashPartitionerTest extends TestLogger {
"group1",
"group2",
new ForwardForConsecutiveHashPartitioner<>(
- new KeyGroupStreamPartitioner<>(record -> 0L, 100)));
+ new KeyGroupStreamPartitioner<>(record -> 0L, 100)),
+ streamExchangeMode);
List<JobVertex> jobVertices = jobGraph.getVerticesSortedTopologicallyFromSources();
Assertions.assertThat(jobVertices.size()).isEqualTo(2);
JobVertex sourceVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(0);
[flink] 03/03: [FLINK-31055][runtime] Fix the bug that dynamic flag of stream graph does not take effect when translating the transformations
Posted by wa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wanglijie pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 629bc9a3d35f5141324ef4cfa9b255dfcc069d20
Author: Lijie Wang <wa...@gmail.com>
AuthorDate: Thu Feb 16 12:12:17 2023 +0800
[FLINK-31055][runtime] Fix the bug that dynamic flag of stream graph does not take effect when translating the transformations
This closes #21932
---
.../streaming/api/graph/StreamGraphGenerator.java | 21 +++++----
.../ForwardForUnspecifiedPartitionerTest.java | 55 +++++++++++++++++++---
2 files changed, 60 insertions(+), 16 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
index c4f2c579493..3859c2bb9b2 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
@@ -323,15 +323,6 @@ public class StreamGraphGenerator {
setFineGrainedGlobalStreamExchangeMode(streamGraph);
- Optional<JobManagerOptions.SchedulerType> schedulerTypeOptional =
- executionConfig.getSchedulerType();
- boolean dynamic =
- shouldExecuteInBatchMode
- && schedulerTypeOptional.orElse(
- JobManagerOptions.SchedulerType.AdaptiveBatch)
- == JobManagerOptions.SchedulerType.AdaptiveBatch;
- streamGraph.setDynamic(dynamic);
-
for (StreamNode node : streamGraph.getStreamNodes()) {
if (node.getInEdges().stream().anyMatch(this::shouldDisableUnalignedCheckpointing)) {
for (StreamEdge edge : node.getInEdges()) {
@@ -354,6 +345,17 @@ public class StreamGraphGenerator {
return partitioner.isPointwise() || partitioner.isBroadcast();
}
+ private void setDynamic(final StreamGraph graph) {
+ Optional<JobManagerOptions.SchedulerType> schedulerTypeOptional =
+ executionConfig.getSchedulerType();
+ boolean dynamic =
+ shouldExecuteInBatchMode
+ && schedulerTypeOptional.orElse(
+ JobManagerOptions.SchedulerType.AdaptiveBatch)
+ == JobManagerOptions.SchedulerType.AdaptiveBatch;
+ graph.setDynamic(dynamic);
+ }
+
private void configureStreamGraph(final StreamGraph graph) {
checkNotNull(graph);
@@ -368,6 +370,7 @@ public class StreamGraphGenerator {
graph.setEnableCheckpointsAfterTasksFinish(
configuration.get(
ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH));
+ setDynamic(graph);
if (shouldExecuteInBatchMode) {
configureStreamGraphBatch(graph);
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/ForwardForUnspecifiedPartitionerTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/ForwardForUnspecifiedPartitionerTest.java
index 1f736ad19de..1e49529dc1c 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/ForwardForUnspecifiedPartitionerTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/ForwardForUnspecifiedPartitionerTest.java
@@ -17,20 +17,25 @@
package org.apache.flink.streaming.runtime.partitioner;
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
import org.apache.flink.streaming.api.graph.NonChainedOutput;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.graph.StreamEdge;
-import org.apache.flink.util.TestLogger;
-import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.List;
+import static org.assertj.core.api.Assertions.assertThat;
+
/** Test for {@link ForwardForUnspecifiedPartitioner}. */
-class ForwardForUnspecifiedPartitionerTest extends TestLogger {
+class ForwardForUnspecifiedPartitionerTest {
@Test
void testConvertToForwardPartitioner() {
@@ -38,12 +43,12 @@ class ForwardForUnspecifiedPartitionerTest extends TestLogger {
StreamPartitionerTestUtils.createJobGraph(
"group1", "group1", new ForwardForUnspecifiedPartitioner<>());
List<JobVertex> jobVertices = jobGraph.getVerticesSortedTopologicallyFromSources();
- Assertions.assertThat(jobVertices.size()).isEqualTo(1);
+ assertThat(jobVertices.size()).isEqualTo(1);
JobVertex vertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(0);
StreamConfig sourceConfig = new StreamConfig(vertex.getConfiguration());
StreamEdge edge = sourceConfig.getChainedOutputs(getClass().getClassLoader()).get(0);
- Assertions.assertThat(edge.getPartitioner()).isInstanceOf(ForwardPartitioner.class);
+ assertThat(edge.getPartitioner()).isInstanceOf(ForwardPartitioner.class);
}
@Test
@@ -52,12 +57,48 @@ class ForwardForUnspecifiedPartitionerTest extends TestLogger {
StreamPartitionerTestUtils.createJobGraph(
"group1", "group2", new ForwardForUnspecifiedPartitioner<>());
List<JobVertex> jobVertices = jobGraph.getVerticesSortedTopologicallyFromSources();
- Assertions.assertThat(jobVertices.size()).isEqualTo(2);
+ assertThat(jobVertices.size()).isEqualTo(2);
JobVertex sourceVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(0);
StreamConfig sourceConfig = new StreamConfig(sourceVertex.getConfiguration());
NonChainedOutput output =
sourceConfig.getOperatorNonChainedOutputs(getClass().getClassLoader()).get(0);
- Assertions.assertThat(output.getPartitioner()).isInstanceOf(RescalePartitioner.class);
+ assertThat(output.getPartitioner()).isInstanceOf(RescalePartitioner.class);
+ }
+
+ @Test
+ void testConvertToCorrectPartitioner() {
+ testConvertToCorrectPartitioner(null, RescalePartitioner.class);
+ testConvertToCorrectPartitioner(
+ JobManagerOptions.SchedulerType.AdaptiveBatch, RescalePartitioner.class);
+ testConvertToCorrectPartitioner(
+ JobManagerOptions.SchedulerType.Default, ForwardPartitioner.class);
+ }
+
+ private void testConvertToCorrectPartitioner(
+ JobManagerOptions.SchedulerType scheduler, Class<?> expectedPartitioner) {
+ Configuration configuration = new Configuration();
+ if (scheduler != null) {
+ configuration.set(JobManagerOptions.SCHEDULER, scheduler);
+ }
+ final StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment(configuration);
+ env.setRuntimeMode(RuntimeExecutionMode.BATCH);
+ env.fromSequence(0, 99)
+ .slotSharingGroup("group1")
+ .name("source")
+ .addSink(new DiscardingSink<>())
+ .slotSharingGroup("group2")
+ .name("sink");
+
+ List<JobVertex> jobVertices =
+ env.getStreamGraph().getJobGraph().getVerticesSortedTopologicallyFromSources();
+ assertThat(jobVertices.size()).isEqualTo(2);
+
+ JobVertex sourceVertex = jobVertices.get(0);
+ StreamConfig sourceConfig = new StreamConfig(sourceVertex.getConfiguration());
+ NonChainedOutput output =
+ sourceConfig.getOperatorNonChainedOutputs(getClass().getClassLoader()).get(0);
+ assertThat(output.getPartitioner()).isInstanceOf(expectedPartitioner);
}
}