[BEAM-5464] Use BATCH_FORCED as the default ExecutionMode for batch pipeline #6897
Conversation
| @Description( | ||
| "Flink mode for data exchange for batch pipeline. " | ||
| + "Reference {@link org.apache.flink.api.common.ExecutionMode}") | ||
| @Default.Enum("BATCH_FORCED") |
There was a problem hiding this comment.
I'm a bit hesitant to change the default value here. This is also used by the non-portable FlinkRunner and the default is PIPELINED. We haven't heard from anyone having issues with the batch execution. I'd leave this at the Flink default until we have found out the exact issue.
There was a problem hiding this comment.
Also ok for me to add a link to a JIRA issue to further investigate this issue.
There was a problem hiding this comment.
Sounds good, I will update the default.
Flink Jira for reference https://issues.apache.org/jira/browse/FLINK-10672
| "Flink mode for data exchange for batch pipeline. " | ||
| + "Reference {@link org.apache.flink.api.common.ExecutionMode}") | ||
| @Default.Enum("BATCH_FORCED") | ||
| ExecutionMode getExecutionModeForBatch(); |
There was a problem hiding this comment.
Can we add a test in PipelineOptionsTest?
There was a problem hiding this comment.
TheDefault Enum test is done in ProxyInvocationHandlerTest.java
Please let me know if you are referring to some other test.
There was a problem hiding this comment.
I meant PipelineOptionsTest to test the default values. I'm adding it in the merge commit.
There was a problem hiding this comment.
Should this be getFlinkExecutionModeForBatch, as it seems rather Flink-specific?
There was a problem hiding this comment.
Considering the global namespace of the option names that would be preferred. However, not a single other option so far has a Flink prefix. Perhaps we have to come up with a way to scope option names?
angoenka
force-pushed
the
tfx_add_execution_mode_option
branch
from
514f46a
to
329e51f
Compare
| "Flink mode for data exchange for batch pipeline. " | ||
| + "Reference {@link org.apache.flink.api.common.ExecutionMode}") | ||
| @Default.Enum("BATCH_FORCED") | ||
| ExecutionMode getExecutionModeForBatch(); |
There was a problem hiding this comment.
Should this be getFlinkExecutionModeForBatch, as it seems rather Flink-specific?
Yes, it just enables to set it. The default remains unchanged ( |
Use BATCH_FORCED ExecutionMode for Flink batch pipelines to avoid flink scheduling dead lock.
Beam merges the chained ProcessBundles in ExecutionStage so this should not have a lot of over head.
Also introducing a parameter to change batch execution mode if needed.
Reference upstream jira https://issues.apache.org/jira/browse/FLINK-10672
Please add a meaningful description for your change here
Follow this checklist to help us incorporate your contribution quickly and easily:
[BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replaceBEAM-XXXwith the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.It will help us expedite review of your Pull Request if you tag someone (e.g.
@username) to look at it.Post-Commit Tests Status (on master branch)