[BEAM-5520] Flink pipeline option to run SDK harness per subtask. #6524
Conversation
|
R: @angoenka |
|
Added server option to control default behavior. When the pipeline option value is |
|
@mxm this will also take care of https://issues.apache.org/jira/browse/BEAM-5516 |
tweise
force-pushed
the
BEAM-5520.perOperatorJobBundleFactory
branch
from
a1000c3
to
9147e0e
Compare
| } | ||
|
|
||
| @Option(name = "--sdk-worker-parallelism", usage = "Parallelism of SDK worker processes") | ||
| private String sdkWorkerParallelism = "[pipeline]"; |
There was a problem hiding this comment.
Please document what the setting configures and what the default value does.
| @@ -31,6 +31,8 @@ | |||
| public interface FlinkPipelineOptions | |||
| extends PipelineOptions, ApplicationNameOptions, StreamingOptions { | |||
|
|
|||
| String AUTO = "[auto]"; | |||
There was a problem hiding this comment.
Should we also add the other options here? Or make this an enum?
There was a problem hiding this comment.
Did not check if/how enum is supported, but this constant is now only used for the master URL. I think it might be good to get rid of the [auto] value for that option as well and just use null instead. As separate change though.
| + "'[pipeline]' (single SDK harness process per pipeline and task manager JVM) or " | ||
| + "'[stage]' (separate SDK harness for every executable stage.)") | ||
| @Default.String(AUTO) | ||
| String getSdkWorkerParallelism(); |
There was a problem hiding this comment.
I think this change is great but I wonder whether we should expose this as a general PortablePipelineOptions?
| static Factory factory() { | ||
| return FlinkDefaultExecutableStageContext.ReferenceCountingFactory.REFERENCE_COUNTING; | ||
| static Factory factory(FlinkPipelineOptions options) { | ||
| if ("[stage]".equals(options.getSdkWorkerParallelism())) { |
There was a problem hiding this comment.
Perhaps define this constant in the Options?
| @@ -141,8 +144,17 @@ public void mapPartition( | |||
|
|
|||
| @Override | |||
| public void close() throws Exception { | |||
| try (AutoCloseable bundleFactoryCloser = stageBundleFactory) {} | |||
| try (AutoCloseable closable = stageContext) {} | |||
| // close may be called multiple times when an exception is thrown | |||
There was a problem hiding this comment.
Should we synchronize close then? The calls could be interleaved.
There was a problem hiding this comment.
It's not called concurrently, it is just being called again when the first call throws an exception.
| @@ -171,7 +171,9 @@ public StageBundleFactory forStage(ExecutableStage executableStage) { | |||
|
|
|||
| @Override | |||
| public void close() throws Exception { | |||
|
|
|||
tweise
force-pushed
the
BEAM-5520.perOperatorJobBundleFactory
branch
from
0b54f21
to
3a5d772
Compare
tweise
force-pushed
the
BEAM-5520.perOperatorJobBundleFactory
branch
from
3a5d772
to
1fdc2f9
Compare
Flink pipeline option to run SDK harness per subtask.
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)