You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "JunRuiLee (via GitHub)" <gi...@apache.org> on 2023/01/28 15:24:18 UTC
[GitHub] [flink] JunRuiLee opened a new pull request, #21772: Flink 30684
JunRuiLee opened a new pull request, #21772:
URL: https://github.com/apache/flink/pull/21772
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
This change proposes to add a parallelismConfigured property as a flag to identify whether the parallelism of node is used "parallelism.default" or not. If the vertex's parallelismConfigured is true, the AdaptiveBatchScheduler will not automatically deciding parallelisms for it. Otherwise, AdaptiveBatchScheduler will automatically deciding parallelisms and use the "parallelism.default" as an alternative value for the "jobmanager.adaptive-batch-scheduler.max-parallelism".
This change will make user do not need to configure "parallelism.default" as "-1" to automatically deciding parallelisms for vertices.
## Brief change log
- Introduce parallelismConfigured as a flag which operator used default parallelism and can automatically derive parallelism.
## Verifying this change
This change is already covered by existing tests
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not documented
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JunRuiLee commented on pull request #21772: [FLINK-30684][runtime] Use the default parallelism as a flag for vertices that can automatically derive parallelism.
Posted by "JunRuiLee (via GitHub)" <gi...@apache.org>.
JunRuiLee commented on PR #21772:
URL: https://github.com/apache/flink/pull/21772#issuecomment-1409636657
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zhuzhurk commented on a diff in pull request #21772: [FLINK-30684][runtime] Use the default parallelism as a flag for vertices that can automatically derive parallelism.
Posted by "zhuzhurk (via GitHub)" <gi...@apache.org>.
zhuzhurk commented on code in PR #21772:
URL: https://github.com/apache/flink/pull/21772#discussion_r1090405433
##########
flink-optimizer/src/main/java/org/apache/flink/optimizer/plantranslate/JobGraphGenerator.java:
##########
@@ -591,6 +605,9 @@ public void postVisit(PlanNode node) {
.addOutputShipStrategy(ShipStrategyType.FORWARD);
}
chainedTask.setContainingVertex(container);
+ if (!parallelismConfigured) {
Review Comment:
```suggestion
if (!node.isParallelismConfigured()) {
```
And remove the above `parallelismConfigured`.
##########
flink-optimizer/src/main/java/org/apache/flink/optimizer/traversals/GraphCreatingVisitor.java:
##########
@@ -241,17 +241,19 @@ public boolean preVisit(Operator<?> c) {
// Keep parallelism of union undefined for now.
// It will be determined based on the parallelism of its successor.
par = -1;
+ n.setParallelismConfigured(false);
} else if (par > 0) {
if (this.forceParallelism && par != this.defaultParallelism) {
par = this.defaultParallelism;
+ n.setParallelismConfigured(false);
Review Comment:
This is a forced parallelism and should not be changed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JunRuiLee commented on pull request #21772: [FLINK-30684][runtime] Use the default parallelism as a flag for vertices that can automatically derive parallelism.
Posted by "JunRuiLee (via GitHub)" <gi...@apache.org>.
JunRuiLee commented on PR #21772:
URL: https://github.com/apache/flink/pull/21772#issuecomment-1409616752
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JunRuiLee commented on pull request #21772: [FLINK-30684][runtime] Use the default parallelism as a flag for vertices that can automatically derive parallelism.
Posted by "JunRuiLee (via GitHub)" <gi...@apache.org>.
JunRuiLee commented on PR #21772:
URL: https://github.com/apache/flink/pull/21772#issuecomment-1407899106
@zhuzhurk I have addressed all comments, please take a look, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zhuzhurk closed pull request #21772: [FLINK-30684][runtime] Use the default parallelism as a flag for vertices that can automatically derive parallelism.
Posted by "zhuzhurk (via GitHub)" <gi...@apache.org>.
zhuzhurk closed pull request #21772: [FLINK-30684][runtime] Use the default parallelism as a flag for vertices that can automatically derive parallelism.
URL: https://github.com/apache/flink/pull/21772
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zhuzhurk commented on pull request #21772: [FLINK-30684][runtime] Use the default parallelism as a flag for vertices that can automatically derive parallelism.
Posted by "zhuzhurk (via GitHub)" <gi...@apache.org>.
zhuzhurk commented on PR #21772:
URL: https://github.com/apache/flink/pull/21772#issuecomment-1408508734
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zhuzhurk commented on a diff in pull request #21772: [FLINK-30684][runtime] Use the default parallelism as a flag for vertices that can automatically derive parallelism.
Posted by "zhuzhurk (via GitHub)" <gi...@apache.org>.
zhuzhurk commented on code in PR #21772:
URL: https://github.com/apache/flink/pull/21772#discussion_r1089997741
##########
flink-core/src/main/java/org/apache/flink/api/dag/Transformation.java:
##########
@@ -111,6 +112,8 @@
// This is used to assign a unique ID to every Transformation
private static final AtomicInteger ID_COUNTER = new AtomicInteger(0);
+ private boolean parallelismConfigured;
Review Comment:
It's better to add some comments to explain what it is used for. e.g. "If true, the parallelism of the transformation is explicitly set and should be respected. Otherwise the parallelism can be changed at runtime."
##########
flink-optimizer/src/main/java/org/apache/flink/optimizer/plantranslate/JobGraphGenerator.java:
##########
@@ -430,6 +435,7 @@ public boolean preVisit(PlanNode node) {
// set parallelism
int pd = node.getParallelism();
vertex.setParallelism(pd);
+ vertex.setParallelismConfigured(node.isParallelismConfigured());
Review Comment:
Looks to me the chained tasks(nodes) are not considered?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java:
##########
@@ -878,6 +879,10 @@ private StreamConfig createJobVertex(Integer streamNodeId, OperatorChainInfo cha
int parallelism = streamNode.getParallelism();
+ if (streamGraph.isDynamic() && !streamNode.isParallelismConfigured()) {
+ streamNode.setParallelism(ExecutionConfig.PARALLELISM_DEFAULT, false);
Review Comment:
Why is this needed? In other words, would the `jobVertex.setParallelism()` work on its own?
What I'm concerned here is whether this change will lead to a different chaining result.
##########
flink-core/src/main/java/org/apache/flink/configuration/ExecutionBatchOptions.java:
##########
@@ -0,0 +1,19 @@
+package org.apache.flink.configuration;
+
+import org.apache.flink.annotation.docs.Documentation;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/** Configuration options for the batch execution. */
+public class ExecutionBatchOptions {
+
+ @Documentation.Section({
+ Documentation.Sections.EXPERT_SCHEDULING,
+ Documentation.Sections.ALL_JOB_MANAGER
Review Comment:
I think adding it to EXPERT_SCHEDULING section is enough.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java:
##########
@@ -755,6 +755,12 @@ public boolean isDynamic() {
return dynamic;
}
+ public void setParallelism(Integer vertexID, int parallelism, boolean parallelismConfigured) {
+ if (getStreamNode(vertexID) != null) {
Review Comment:
vertexID -> vertexId
##########
flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/OptimizerNode.java:
##########
@@ -407,6 +409,14 @@ public void setParallelism(int parallelism) {
this.parallelism = parallelism;
}
+ public boolean isParallelismConfigured() {
+ return parallelismConfigured;
+ }
+
+ public void enableParallelismConfigured(boolean parallelismConfigured) {
Review Comment:
enableParallelismConfigured -> setParallelismConfigured
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java:
##########
@@ -254,6 +253,46 @@ void testEnabledUnalignedCheckAndDisabledCheckpointing() {
assertThat(streamConfig.isUnalignedCheckpointsEnabled()).isFalse();
}
+ @Test
+ public void testTransformationSetParallelism() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.fromSequence(1L, 3L).map(i -> i).setParallelism(10).print().setParallelism(20);
+ StreamGraph streamGraph = env.getStreamGraph();
+
+ // check the streamGraph parallleism configured
+ StreamNode[] streamNodes = streamGraph.getStreamNodes().toArray(new StreamNode[0]);
+ assertThat(streamNodes[0].isParallelismConfigured()).isFalse();
+ assertThat(streamNodes[1].isParallelismConfigured()).isTrue();
+ assertThat(streamNodes[2].isParallelismConfigured()).isTrue();
+
+ // check the jobGraph parallelism configured
+ JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
+ List<JobVertex> vertices = jobGraph.getVerticesSortedTopologicallyFromSources();
+ assertThat(jobGraph.getNumberOfVertices()).isEqualTo(3);
+ assertThat(vertices.get(0).isParallelismConfigured()).isFalse();
+ assertThat(vertices.get(1).isParallelismConfigured()).isTrue();
+ assertThat(vertices.get(2).isParallelismConfigured()).isTrue();
+ }
+
Review Comment:
It's better to add a test to check that for dynamic graph, the vertex parallelism is set to -1, while for non-dynamic graph, the parallelism is the default parallelism.
##########
flink-core/src/main/java/org/apache/flink/api/dag/Transformation.java:
##########
@@ -183,11 +186,31 @@ public static int getNewNodeId() {
* @param parallelism The parallelism of this {@code Transformation}
*/
public Transformation(String name, TypeInformation<T> outputType, int parallelism) {
+ this(name, outputType, parallelism, true);
+ }
+
+ /**
+ * Creates a new {@code Transformation} with the given name, output type and parallelism.
+ *
+ * @param name The name of the {@code Transformation}, this will be shown in Visualizations and
+ * the Log
+ * @param outputType The output type of this {@code Transformation}
+ * @param parallelism The parallelism of this {@code Transformation}
+ * @param parallelismConfigured if true, the parallelism of transformation is not use default
+ * parallelism
Review Comment:
Maybe "If true, the parallelism of the transformation is explicitly set and should be respected. Otherwise the parallelism can be changed at runtime."
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchSchedulerFactory.java:
##########
@@ -168,6 +169,17 @@ public SchedulerNG createInstance(
new VertexwiseSchedulingStrategy.Factory(
loadInputConsumableDeciderFactory(hybridPartitionDataConsumeConstraint));
+ int defaultMaxParallelism =
+ jobMasterConfiguration
+ .getOptional(JobManagerOptions.ADAPTIVE_BATCH_SCHEDULER_MAX_PARALLELISM)
+ .orElse(
+ jobMasterConfiguration
+ .getOptional(CoreOptions.DEFAULT_PARALLELISM)
+ .orElse(
+ jobMasterConfiguration.getInteger(
+ JobManagerOptions
+ .ADAPTIVE_BATCH_SCHEDULER_MAX_PARALLELISM)));
Review Comment:
It's more clear to directly use `ADAPTIVE_BATCH_SCHEDULER_MAX_PARALLELISM.defaultValue()` here.
##########
flink-core/src/main/java/org/apache/flink/configuration/ExecutionBatchOptions.java:
##########
@@ -0,0 +1,19 @@
+package org.apache.flink.configuration;
+
+import org.apache.flink.annotation.docs.Documentation;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/** Configuration options for the batch execution. */
Review Comment:
the batch execution -> batch job execution
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamNode.java:
##########
@@ -191,7 +193,12 @@ public int getParallelism() {
}
public void setParallelism(Integer parallelism) {
+ setParallelism(parallelism, true);
+ }
+
+ void setParallelism(Integer parallelism, boolean parallelismConfigured) {
this.parallelism = parallelism;
+ this.parallelismConfigured = parallelismConfigured;
Review Comment:
if `parallelism < 0`, `parallelismConfigured` should be false.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java:
##########
@@ -1861,10 +1880,16 @@ private OperatorID addNodeToChain(int currentNodeId, String operatorName) {
operatorHashes.add(new Tuple2<>(primaryHashBytes, legacyHash.get(currentNodeId)));
}
- streamGraph
- .getStreamNode(currentNodeId)
+ streamNode
.getCoordinatorProvider(operatorName, new OperatorID(getHash(currentNodeId)))
.map(coordinatorProviders::add);
+
+ if (streamGraph.isDynamic()
+ && !streamNode.isParallelismConfigured()
+ && startNodeId != currentNodeId) {
+ streamNode.setParallelism(ExecutionConfig.PARALLELISM_DEFAULT, false);
Review Comment:
Why do we need this change?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java:
##########
@@ -141,6 +141,8 @@ public class StreamGraph implements Pipeline {
private boolean dynamic;
+ private boolean autoParallelism;
Review Comment:
autoParallelism -> autoParallelismEnabled
##########
flink-core/src/main/java/org/apache/flink/configuration/ExecutionBatchOptions.java:
##########
@@ -0,0 +1,19 @@
+package org.apache.flink.configuration;
+
+import org.apache.flink.annotation.docs.Documentation;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/** Configuration options for the batch execution. */
+public class ExecutionBatchOptions {
Review Comment:
I prefer to name it as `BatchExecutionOptions`.
##########
flink-core/src/main/java/org/apache/flink/configuration/ExecutionBatchOptions.java:
##########
@@ -0,0 +1,19 @@
+package org.apache.flink.configuration;
+
+import org.apache.flink.annotation.docs.Documentation;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/** Configuration options for the batch execution. */
+public class ExecutionBatchOptions {
+
+ @Documentation.Section({
+ Documentation.Sections.EXPERT_SCHEDULING,
+ Documentation.Sections.ALL_JOB_MANAGER
+ })
+ public static final ConfigOption<Boolean> ADAPTIVE_AUTO_PARALLELISM_ENABLED =
+ key("execution.batch.adaptive.auto-parallelism.enabled")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription("Controls whether to enable auto parallelism derivation.");
Review Comment:
-> If true, Flink will automatically decide the parallelism of operators in batch jobs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JunRuiLee commented on pull request #21772: [FLINK-30684][runtime] Use the default parallelism as a flag for vertices that can automatically derive parallelism.
Posted by "JunRuiLee (via GitHub)" <gi...@apache.org>.
JunRuiLee commented on PR #21772:
URL: https://github.com/apache/flink/pull/21772#issuecomment-1409575291
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JunRuiLee commented on pull request #21772: [FLINK-30684][runtime] Use the default parallelism as a flag for vertices that can automatically derive parallelism.
Posted by "JunRuiLee (via GitHub)" <gi...@apache.org>.
JunRuiLee commented on PR #21772:
URL: https://github.com/apache/flink/pull/21772#issuecomment-1408821831
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #21772: Flink 30684
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #21772:
URL: https://github.com/apache/flink/pull/21772#issuecomment-1407422894
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "477bfc757c683af6ca230d881f57232061c4fc55",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "477bfc757c683af6ca230d881f57232061c4fc55",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 477bfc757c683af6ca230d881f57232061c4fc55 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JunRuiLee commented on pull request #21772: [FLINK-30684][runtime] Use the default parallelism as a flag for vertices that can automatically derive parallelism.
Posted by "JunRuiLee (via GitHub)" <gi...@apache.org>.
JunRuiLee commented on PR #21772:
URL: https://github.com/apache/flink/pull/21772#issuecomment-1407551721
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zhuzhurk commented on a diff in pull request #21772: [FLINK-30684][runtime] Use the default parallelism as a flag for vertices that can automatically derive parallelism.
Posted by "zhuzhurk (via GitHub)" <gi...@apache.org>.
zhuzhurk commented on code in PR #21772:
URL: https://github.com/apache/flink/pull/21772#discussion_r1090413159
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java:
##########
@@ -1839,6 +1839,7 @@ private OperatorChainInfo(
this.streamGraph = streamGraph;
this.parallelismConfigured =
streamGraph.getStreamNode(startNodeId).isParallelismConfigured();
+ this.chainedNodes = new ArrayList<>();
Review Comment:
The headNode should also be included.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zhuzhurk commented on a diff in pull request #21772: [FLINK-30684][runtime] Use the default parallelism as a flag for vertices that can automatically derive parallelism.
Posted by "zhuzhurk (via GitHub)" <gi...@apache.org>.
zhuzhurk commented on code in PR #21772:
URL: https://github.com/apache/flink/pull/21772#discussion_r1090206907
##########
flink-optimizer/src/main/java/org/apache/flink/optimizer/plantranslate/JobGraphGenerator.java:
##########
@@ -565,6 +589,7 @@ public void postVisit(PlanNode node) {
if (container == null) {
final PlanNode sourceNode = inConn.getSource();
+ boolean parallelismConfigured = sourceNode.isParallelismConfigured();
Review Comment:
`parallelismConfigured` should be from the `node` instead of the `sourceNode`.
Could we have some tests to verify the `parallelismConfigured` of DataSet jobs?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org