You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "xinyuiscool (via GitHub)" <gi...@apache.org> on 2023/04/18 01:56:33 UTC
[GitHub] [beam] xinyuiscool opened a new pull request, #26314: Clean up bundle config generation for Samza Runner
xinyuiscool opened a new pull request, #26314:
URL: https://github.com/apache/beam/pull/26314
Refactor the samza runner configs into a new function with consideration of autosizing configs. Also add unit test for it.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/get-started-contributing/#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] mynameborat commented on a diff in pull request #26314: Clean up bundle config generation for Samza Runner
Posted by "mynameborat (via GitHub)" <gi...@apache.org>.
mynameborat commented on code in PR #26314:
URL: https://github.com/apache/beam/pull/26314#discussion_r1170536592
##########
runners/samza/src/main/java/org/apache/beam/runners/samza/translation/ConfigBuilder.java:
##########
@@ -131,6 +117,30 @@ public Config build() {
}
}
+ @VisibleForTesting
+ static Map<String, String> createBundleConfig(
+ SamzaPipelineOptions options, Map<String, String> config) {
+ ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
+ builder.put(MAX_CONCURRENCY, String.valueOf(options.getMaxBundleSize()));
+
+ if (options.getMaxBundleSize() > 1) {
+ final int threadPoolSize = ConfigUtils.asJobConfig(config).getThreadPoolSize();
+ LOG.info("Remove threadPoolSize configs when maxBundleSize > 1");
+ builder.put(JOB_CONTAINER_THREAD_POOL_SIZE, "0");
+ builder.put(JOB_AUTOSIZING_CONTAINER_THREAD_POOL_SIZE, "0");
+
+ if (threadPoolSize > 1 && options.getNumThreadsForProcessElement() <= 1) {
+ // In case the user sets the thread pool through samza config instead options,
+ // set the bundle thread pool size based on container thread pool config
+ // this allows Samza auto-sizing to tune the threads
+ LOG.info("Convert threadPoolSize {} to numThreadsForProcessElement", threadPoolSize);
+ // NumThreadsForProcessElement in option is the source of truth
+ options.setNumThreadsForProcessElement(threadPoolSize);
Review Comment:
I feel the inconsistency in behavior is as well confusing to the users. If we want to establish user code as the highest precedence, maybe we should not tie autosizing `job.autosizing.container.thread.pool.size` into this.
Consider this example
1. user configures `job.container.thread.pool.size`, it is reflected in the application behavior
2. autosizing is enabled and user configuration no longer is reflected in the application behavior
3. User sets it in code and the change is reflected in application behavior
user configuration << framework configuration override << user code
Also consider autosizing devs/support team not realizing knobs x, y and z follows auto-sizer values as higher precedence on their end vs a, b follows user values as higher precedence.
>The other problem is that we need a way for user to override without turning off autosizing
I think the recommended way for now is to use auto-sizer tool to overrides specific knobs.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] mynameborat commented on a diff in pull request #26314: Clean up bundle config generation for Samza Runner
Posted by "mynameborat (via GitHub)" <gi...@apache.org>.
mynameborat commented on code in PR #26314:
URL: https://github.com/apache/beam/pull/26314#discussion_r1170412724
##########
runners/samza/src/main/java/org/apache/beam/runners/samza/translation/ConfigBuilder.java:
##########
@@ -131,6 +117,30 @@ public Config build() {
}
}
+ @VisibleForTesting
+ static Map<String, String> createBundleConfig(
+ SamzaPipelineOptions options, Map<String, String> config) {
+ ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
+ builder.put(MAX_CONCURRENCY, String.valueOf(options.getMaxBundleSize()));
+
+ if (options.getMaxBundleSize() > 1) {
+ final int threadPoolSize = ConfigUtils.asJobConfig(config).getThreadPoolSize();
+ LOG.info("Remove threadPoolSize configs when maxBundleSize > 1");
+ builder.put(JOB_CONTAINER_THREAD_POOL_SIZE, "0");
+ builder.put(JOB_AUTOSIZING_CONTAINER_THREAD_POOL_SIZE, "0");
+
+ if (threadPoolSize > 1 && options.getNumThreadsForProcessElement() <= 1) {
+ // In case the user sets the thread pool through samza config instead options,
+ // set the bundle thread pool size based on container thread pool config
+ // this allows Samza auto-sizing to tune the threads
+ LOG.info("Convert threadPoolSize {} to numThreadsForProcessElement", threadPoolSize);
+ // NumThreadsForProcessElement in option is the source of truth
+ options.setNumThreadsForProcessElement(threadPoolSize);
Review Comment:
Shouldn't autosizing take precedence over user configuration? i.e.,
We'd need to also set the threadPoolSize regardless of user's `getNumThreadsForProcessElement()` if autosizing is enabled.
##########
runners/samza/src/main/java/org/apache/beam/runners/samza/translation/ConfigBuilder.java:
##########
@@ -131,6 +117,30 @@ public Config build() {
}
}
+ @VisibleForTesting
+ static Map<String, String> createBundleConfig(
+ SamzaPipelineOptions options, Map<String, String> config) {
+ ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
+ builder.put(MAX_CONCURRENCY, String.valueOf(options.getMaxBundleSize()));
+
+ if (options.getMaxBundleSize() > 1) {
+ final int threadPoolSize = ConfigUtils.asJobConfig(config).getThreadPoolSize();
+ LOG.info("Remove threadPoolSize configs when maxBundleSize > 1");
+ builder.put(JOB_CONTAINER_THREAD_POOL_SIZE, "0");
+ builder.put(JOB_AUTOSIZING_CONTAINER_THREAD_POOL_SIZE, "0");
Review Comment:
Can we add a note on why we reset the value that is set by auto-sizer?
##########
runners/samza/src/test/java/org/apache/beam/runners/samza/translation/ConfigGeneratorTest.java:
##########
@@ -414,4 +417,39 @@ public void processElement(
"TestStoreConfig-1-testState-Same_stateful_ParDo_Name2-changelog",
config2.get("stores.testState-Same_stateful_ParDo_Name2.changelog"));
}
+
+ @Test
+ public void testCreateBundleConfig() {
Review Comment:
Reminder to add another test w/ autosizing enabled vs disabled
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] mynameborat commented on a diff in pull request #26314: Clean up bundle config generation for Samza Runner
Posted by "mynameborat (via GitHub)" <gi...@apache.org>.
mynameborat commented on code in PR #26314:
URL: https://github.com/apache/beam/pull/26314#discussion_r1170458986
##########
runners/samza/src/main/java/org/apache/beam/runners/samza/translation/ConfigBuilder.java:
##########
@@ -131,6 +117,30 @@ public Config build() {
}
}
+ @VisibleForTesting
+ static Map<String, String> createBundleConfig(
+ SamzaPipelineOptions options, Map<String, String> config) {
+ ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
+ builder.put(MAX_CONCURRENCY, String.valueOf(options.getMaxBundleSize()));
+
+ if (options.getMaxBundleSize() > 1) {
+ final int threadPoolSize = ConfigUtils.asJobConfig(config).getThreadPoolSize();
+ LOG.info("Remove threadPoolSize configs when maxBundleSize > 1");
+ builder.put(JOB_CONTAINER_THREAD_POOL_SIZE, "0");
+ builder.put(JOB_AUTOSIZING_CONTAINER_THREAD_POOL_SIZE, "0");
+
+ if (threadPoolSize > 1 && options.getNumThreadsForProcessElement() <= 1) {
+ // In case the user sets the thread pool through samza config instead options,
+ // set the bundle thread pool size based on container thread pool config
+ // this allows Samza auto-sizing to tune the threads
+ LOG.info("Convert threadPoolSize {} to numThreadsForProcessElement", threadPoolSize);
+ // NumThreadsForProcessElement in option is the source of truth
+ options.setNumThreadsForProcessElement(threadPoolSize);
Review Comment:
Isn't this opposite to the existing pattern we have for autosizing? i.e., users can configure resourcing but if they enable autosizing, we treat autosizing overrides as the final values.
This would be inconsistent with the existing behavior for every other auto-sizer tunable configurations.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] xinyuiscool commented on pull request #26314: Clean up bundle config generation for Samza Runner
Posted by "xinyuiscool (via GitHub)" <gi...@apache.org>.
xinyuiscool commented on PR #26314:
URL: https://github.com/apache/beam/pull/26314#issuecomment-1513852693
> @xinyuiscool Can you look into the test failure before you merge?
Yup, seems some env issues in the windows OS test. Will wait till Java PreCommit pass.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] mynameborat commented on pull request #26314: Clean up bundle config generation for Samza Runner
Posted by "mynameborat (via GitHub)" <gi...@apache.org>.
mynameborat commented on PR #26314:
URL: https://github.com/apache/beam/pull/26314#issuecomment-1513850645
@xinyuiscool Can you look into the test failure before you merge?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] jia-gao commented on a diff in pull request #26314: Clean up bundle config generation for Samza Runner
Posted by "jia-gao (via GitHub)" <gi...@apache.org>.
jia-gao commented on code in PR #26314:
URL: https://github.com/apache/beam/pull/26314#discussion_r1170561739
##########
runners/samza/src/main/java/org/apache/beam/runners/samza/translation/ConfigBuilder.java:
##########
@@ -131,6 +117,30 @@ public Config build() {
}
}
+ @VisibleForTesting
+ static Map<String, String> createBundleConfig(
+ SamzaPipelineOptions options, Map<String, String> config) {
+ ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
+ builder.put(MAX_CONCURRENCY, String.valueOf(options.getMaxBundleSize()));
+
+ if (options.getMaxBundleSize() > 1) {
+ final int threadPoolSize = ConfigUtils.asJobConfig(config).getThreadPoolSize();
+ LOG.info("Remove threadPoolSize configs when maxBundleSize > 1");
+ builder.put(JOB_CONTAINER_THREAD_POOL_SIZE, "0");
+ builder.put(JOB_AUTOSIZING_CONTAINER_THREAD_POOL_SIZE, "0");
+
+ if (threadPoolSize > 1 && options.getNumThreadsForProcessElement() <= 1) {
+ // In case the user sets the thread pool through samza config instead options,
+ // set the bundle thread pool size based on container thread pool config
+ // this allows Samza auto-sizing to tune the threads
+ LOG.info("Convert threadPoolSize {} to numThreadsForProcessElement", threadPoolSize);
+ // NumThreadsForProcessElement in option is the source of truth
+ options.setNumThreadsForProcessElement(threadPoolSize);
Review Comment:
To clarify, ` job.container.thread.pool.size` won't work if auto-sizing is enabled.
auto sizing today allows user overrides size configs but still leave auto sizing enabled (asc can still resize the job) by:
1. users can configure (though not recommended) `job.autosizing.container.thread.pool.size`. It will be honored when the job starts
2. when auto-sizing resizes the job, it will submit the job to jetty proxy with a new value of
`job.autosizing.container.thread.pool.size` included in the submission config. and write it to the coordinator
stream
3. If user redeploys the job again, the size in 1 is still honored. If the user removes the config in 1 and redeploy,
the size in 2 will be honored (reading from coordinator stream)
for beam job options like ThreadsForProcessElement, asc is not able to achieve 2 so I agree that in ConfigBuilder we should honor the `job.autosizing.container.thread.pool.size` over options specified in code to be able to resize the job. Afterall, if `job.autosizing.container.thread.pool.size` is present, if means either asc resizes the job or the job owner wants to override the size but still want to leave auto-sizing on
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] xinyuiscool merged pull request #26314: Clean up bundle config generation for Samza Runner
Posted by "xinyuiscool (via GitHub)" <gi...@apache.org>.
xinyuiscool merged PR #26314:
URL: https://github.com/apache/beam/pull/26314
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] xinyuiscool commented on a diff in pull request #26314: Clean up bundle config generation for Samza Runner
Posted by "xinyuiscool (via GitHub)" <gi...@apache.org>.
xinyuiscool commented on code in PR #26314:
URL: https://github.com/apache/beam/pull/26314#discussion_r1170451668
##########
runners/samza/src/main/java/org/apache/beam/runners/samza/translation/ConfigBuilder.java:
##########
@@ -131,6 +117,30 @@ public Config build() {
}
}
+ @VisibleForTesting
+ static Map<String, String> createBundleConfig(
+ SamzaPipelineOptions options, Map<String, String> config) {
+ ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
+ builder.put(MAX_CONCURRENCY, String.valueOf(options.getMaxBundleSize()));
+
+ if (options.getMaxBundleSize() > 1) {
+ final int threadPoolSize = ConfigUtils.asJobConfig(config).getThreadPoolSize();
+ LOG.info("Remove threadPoolSize configs when maxBundleSize > 1");
+ builder.put(JOB_CONTAINER_THREAD_POOL_SIZE, "0");
+ builder.put(JOB_AUTOSIZING_CONTAINER_THREAD_POOL_SIZE, "0");
Review Comment:
Added.
##########
runners/samza/src/test/java/org/apache/beam/runners/samza/translation/ConfigGeneratorTest.java:
##########
@@ -414,4 +417,39 @@ public void processElement(
"TestStoreConfig-1-testState-Same_stateful_ParDo_Name2-changelog",
config2.get("stores.testState-Same_stateful_ParDo_Name2.changelog"));
}
+
+ @Test
+ public void testCreateBundleConfig() {
Review Comment:
Good point. I added the case in the same tests. The results are expected to be the same.
##########
runners/samza/src/main/java/org/apache/beam/runners/samza/translation/ConfigBuilder.java:
##########
@@ -131,6 +117,30 @@ public Config build() {
}
}
+ @VisibleForTesting
+ static Map<String, String> createBundleConfig(
+ SamzaPipelineOptions options, Map<String, String> config) {
+ ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
+ builder.put(MAX_CONCURRENCY, String.valueOf(options.getMaxBundleSize()));
+
+ if (options.getMaxBundleSize() > 1) {
+ final int threadPoolSize = ConfigUtils.asJobConfig(config).getThreadPoolSize();
+ LOG.info("Remove threadPoolSize configs when maxBundleSize > 1");
+ builder.put(JOB_CONTAINER_THREAD_POOL_SIZE, "0");
+ builder.put(JOB_AUTOSIZING_CONTAINER_THREAD_POOL_SIZE, "0");
+
+ if (threadPoolSize > 1 && options.getNumThreadsForProcessElement() <= 1) {
+ // In case the user sets the thread pool through samza config instead options,
+ // set the bundle thread pool size based on container thread pool config
+ // this allows Samza auto-sizing to tune the threads
+ LOG.info("Convert threadPoolSize {} to numThreadsForProcessElement", threadPoolSize);
+ // NumThreadsForProcessElement in option is the source of truth
+ options.setNumThreadsForProcessElement(threadPoolSize);
Review Comment:
If the user sets the pipeline options in code, it should take higher precedence. Otherwise it will be quite confusing. Autosizing will work if the user doesn't set pipeline options.
We set threadPoolSize to 0 regardless.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] jia-gao commented on a diff in pull request #26314: Clean up bundle config generation for Samza Runner
Posted by "jia-gao (via GitHub)" <gi...@apache.org>.
jia-gao commented on code in PR #26314:
URL: https://github.com/apache/beam/pull/26314#discussion_r1170567596
##########
runners/samza/src/main/java/org/apache/beam/runners/samza/translation/ConfigBuilder.java:
##########
@@ -131,6 +117,30 @@ public Config build() {
}
}
+ @VisibleForTesting
+ static Map<String, String> createBundleConfig(
+ SamzaPipelineOptions options, Map<String, String> config) {
+ ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
+ builder.put(MAX_CONCURRENCY, String.valueOf(options.getMaxBundleSize()));
+
+ if (options.getMaxBundleSize() > 1) {
+ final int threadPoolSize = ConfigUtils.asJobConfig(config).getThreadPoolSize();
+ LOG.info("Remove threadPoolSize configs when maxBundleSize > 1");
+ builder.put(JOB_CONTAINER_THREAD_POOL_SIZE, "0");
+ builder.put(JOB_AUTOSIZING_CONTAINER_THREAD_POOL_SIZE, "0");
+
+ if (threadPoolSize > 1 && options.getNumThreadsForProcessElement() <= 1) {
+ // In case the user sets the thread pool through samza config instead options,
+ // set the bundle thread pool size based on container thread pool config
+ // this allows Samza auto-sizing to tune the threads
+ LOG.info("Convert threadPoolSize {} to numThreadsForProcessElement", threadPoolSize);
+ // NumThreadsForProcessElement in option is the source of truth
+ options.setNumThreadsForProcessElement(threadPoolSize);
Review Comment:
> I think the recommended way for now is to use auto-sizer tool to overrides specific knobs.
asc-tool does the same thing as step 2 above. ConfigBuilder still needs to address the problem job config vs pipeline option precedence here
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] github-actions[bot] commented on pull request #26314: Clean up bundle config generation for Samza Runner
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #26314:
URL: https://github.com/apache/beam/pull/26314#issuecomment-1512327287
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] xinyuiscool commented on pull request #26314: Clean up bundle config generation for Samza Runner
Posted by "xinyuiscool (via GitHub)" <gi...@apache.org>.
xinyuiscool commented on PR #26314:
URL: https://github.com/apache/beam/pull/26314#issuecomment-1512326423
R: @mynameborat please review. 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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] jia-gao commented on a diff in pull request #26314: Clean up bundle config generation for Samza Runner
Posted by "jia-gao (via GitHub)" <gi...@apache.org>.
jia-gao commented on code in PR #26314:
URL: https://github.com/apache/beam/pull/26314#discussion_r1170567596
##########
runners/samza/src/main/java/org/apache/beam/runners/samza/translation/ConfigBuilder.java:
##########
@@ -131,6 +117,30 @@ public Config build() {
}
}
+ @VisibleForTesting
+ static Map<String, String> createBundleConfig(
+ SamzaPipelineOptions options, Map<String, String> config) {
+ ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
+ builder.put(MAX_CONCURRENCY, String.valueOf(options.getMaxBundleSize()));
+
+ if (options.getMaxBundleSize() > 1) {
+ final int threadPoolSize = ConfigUtils.asJobConfig(config).getThreadPoolSize();
+ LOG.info("Remove threadPoolSize configs when maxBundleSize > 1");
+ builder.put(JOB_CONTAINER_THREAD_POOL_SIZE, "0");
+ builder.put(JOB_AUTOSIZING_CONTAINER_THREAD_POOL_SIZE, "0");
+
+ if (threadPoolSize > 1 && options.getNumThreadsForProcessElement() <= 1) {
+ // In case the user sets the thread pool through samza config instead options,
+ // set the bundle thread pool size based on container thread pool config
+ // this allows Samza auto-sizing to tune the threads
+ LOG.info("Convert threadPoolSize {} to numThreadsForProcessElement", threadPoolSize);
+ // NumThreadsForProcessElement in option is the source of truth
+ options.setNumThreadsForProcessElement(threadPoolSize);
Review Comment:
> I think the recommended way for now is to use auto-sizer tool to overrides specific knobs.
asc-tool does the same thing as step 2 above. ConfigBuilder still needs to address the problem job config vs pipeline option precedence here
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] mynameborat commented on a diff in pull request #26314: Clean up bundle config generation for Samza Runner
Posted by "mynameborat (via GitHub)" <gi...@apache.org>.
mynameborat commented on code in PR #26314:
URL: https://github.com/apache/beam/pull/26314#discussion_r1170613204
##########
runners/samza/src/main/java/org/apache/beam/runners/samza/translation/ConfigBuilder.java:
##########
@@ -131,6 +117,30 @@ public Config build() {
}
}
+ @VisibleForTesting
+ static Map<String, String> createBundleConfig(
+ SamzaPipelineOptions options, Map<String, String> config) {
+ ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
+ builder.put(MAX_CONCURRENCY, String.valueOf(options.getMaxBundleSize()));
+
+ if (options.getMaxBundleSize() > 1) {
+ final int threadPoolSize = ConfigUtils.asJobConfig(config).getThreadPoolSize();
+ LOG.info("Remove threadPoolSize configs when maxBundleSize > 1");
+ builder.put(JOB_CONTAINER_THREAD_POOL_SIZE, "0");
+ builder.put(JOB_AUTOSIZING_CONTAINER_THREAD_POOL_SIZE, "0");
+
+ if (threadPoolSize > 1 && options.getNumThreadsForProcessElement() <= 1) {
+ // In case the user sets the thread pool through samza config instead options,
+ // set the bundle thread pool size based on container thread pool config
+ // this allows Samza auto-sizing to tune the threads
+ LOG.info("Convert threadPoolSize {} to numThreadsForProcessElement", threadPoolSize);
+ // NumThreadsForProcessElement in option is the source of truth
+ options.setNumThreadsForProcessElement(threadPoolSize);
Review Comment:
We sync'd up offline on this. The viewpoint to have autosizing enabled plus allow user overrides for knobs that autosizing hasn't caught up on is very valid scenario.
We don't have such compartmentalization today as there are relationships between policies dealing with different resources and hence disabling autosizing all out is the current recommendation to override. That said, unblocking this PR for site-up and we can follow up on this
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] mynameborat commented on a diff in pull request #26314: Clean up bundle config generation for Samza Runner
Posted by "mynameborat (via GitHub)" <gi...@apache.org>.
mynameborat commented on code in PR #26314:
URL: https://github.com/apache/beam/pull/26314#discussion_r1170536592
##########
runners/samza/src/main/java/org/apache/beam/runners/samza/translation/ConfigBuilder.java:
##########
@@ -131,6 +117,30 @@ public Config build() {
}
}
+ @VisibleForTesting
+ static Map<String, String> createBundleConfig(
+ SamzaPipelineOptions options, Map<String, String> config) {
+ ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
+ builder.put(MAX_CONCURRENCY, String.valueOf(options.getMaxBundleSize()));
+
+ if (options.getMaxBundleSize() > 1) {
+ final int threadPoolSize = ConfigUtils.asJobConfig(config).getThreadPoolSize();
+ LOG.info("Remove threadPoolSize configs when maxBundleSize > 1");
+ builder.put(JOB_CONTAINER_THREAD_POOL_SIZE, "0");
+ builder.put(JOB_AUTOSIZING_CONTAINER_THREAD_POOL_SIZE, "0");
+
+ if (threadPoolSize > 1 && options.getNumThreadsForProcessElement() <= 1) {
+ // In case the user sets the thread pool through samza config instead options,
+ // set the bundle thread pool size based on container thread pool config
+ // this allows Samza auto-sizing to tune the threads
+ LOG.info("Convert threadPoolSize {} to numThreadsForProcessElement", threadPoolSize);
+ // NumThreadsForProcessElement in option is the source of truth
+ options.setNumThreadsForProcessElement(threadPoolSize);
Review Comment:
I feel the inconsistency in behavior is as well confusing to the users. If we want to establish user code as the highest precedence, maybe we should not tie autosizing `job.autosizing.container.thread.pool.size` into this.
Consider this example
1. user configures `job.container.thread.pool.size`, it is reflected in the application behavior
2. autosizing is enabled and user configuration no longer is reflected in the application behavior
3. User sets it in code and the change is reflected in application behavior
user configuration << framework configuration override << user code
Also consider autosizing devs/support team not realizing knobs x, y and z follows auto-sizer values as higher precedence on their end vs a, b follows user values as higher precedence.
>The other problem is that we need a way for user to override without turning off autosizing
I think the recommended way for now is to use auto-sizer tool to overrides specific knobs.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] jia-gao commented on a diff in pull request #26314: Clean up bundle config generation for Samza Runner
Posted by "jia-gao (via GitHub)" <gi...@apache.org>.
jia-gao commented on code in PR #26314:
URL: https://github.com/apache/beam/pull/26314#discussion_r1170561739
##########
runners/samza/src/main/java/org/apache/beam/runners/samza/translation/ConfigBuilder.java:
##########
@@ -131,6 +117,30 @@ public Config build() {
}
}
+ @VisibleForTesting
+ static Map<String, String> createBundleConfig(
+ SamzaPipelineOptions options, Map<String, String> config) {
+ ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
+ builder.put(MAX_CONCURRENCY, String.valueOf(options.getMaxBundleSize()));
+
+ if (options.getMaxBundleSize() > 1) {
+ final int threadPoolSize = ConfigUtils.asJobConfig(config).getThreadPoolSize();
+ LOG.info("Remove threadPoolSize configs when maxBundleSize > 1");
+ builder.put(JOB_CONTAINER_THREAD_POOL_SIZE, "0");
+ builder.put(JOB_AUTOSIZING_CONTAINER_THREAD_POOL_SIZE, "0");
+
+ if (threadPoolSize > 1 && options.getNumThreadsForProcessElement() <= 1) {
+ // In case the user sets the thread pool through samza config instead options,
+ // set the bundle thread pool size based on container thread pool config
+ // this allows Samza auto-sizing to tune the threads
+ LOG.info("Convert threadPoolSize {} to numThreadsForProcessElement", threadPoolSize);
+ // NumThreadsForProcessElement in option is the source of truth
+ options.setNumThreadsForProcessElement(threadPoolSize);
Review Comment:
To clarify, ` job.container.thread.pool.size` won't work if auto-sizing is enabled.
auto sizing today allows overrides size configs but still leave auto sizing enabled (asc can still resize the job) by:
1. users can configure (though not recommended) `job.autosizing.container.thread.pool.size`. It will be honored when the job starts
2. when auto-sizing resizes the job, it will submit the job to jetty proxy with a new value of
`job.autosizing.container.thread.pool.size` included in the submission config. and write it to the coordinator
stream
3. If user redeploys the job again, the size in 1 is still honored. If the user removes the config in 1 and redeploy,
the size in 2 will be honored (reading from coordinator stream)
for beam job options like ThreadsForProcessElement, asc is not able to achieve 2 so I agree that in ConfigBuilder we should honor the `job.autosizing.container.thread.pool.size` over options specified in code to be able to resize the job. Afterall, if `job.autosizing.container.thread.pool.size` is present, if means either asc resizes the job or the job owner wants to override the size but still want to leave auto-sizing on
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] xinyuiscool commented on a diff in pull request #26314: Clean up bundle config generation for Samza Runner
Posted by "xinyuiscool (via GitHub)" <gi...@apache.org>.
xinyuiscool commented on code in PR #26314:
URL: https://github.com/apache/beam/pull/26314#discussion_r1170609096
##########
runners/samza/src/main/java/org/apache/beam/runners/samza/translation/ConfigBuilder.java:
##########
@@ -131,6 +117,30 @@ public Config build() {
}
}
+ @VisibleForTesting
+ static Map<String, String> createBundleConfig(
+ SamzaPipelineOptions options, Map<String, String> config) {
+ ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
+ builder.put(MAX_CONCURRENCY, String.valueOf(options.getMaxBundleSize()));
+
+ if (options.getMaxBundleSize() > 1) {
+ final int threadPoolSize = ConfigUtils.asJobConfig(config).getThreadPoolSize();
+ LOG.info("Remove threadPoolSize configs when maxBundleSize > 1");
+ builder.put(JOB_CONTAINER_THREAD_POOL_SIZE, "0");
+ builder.put(JOB_AUTOSIZING_CONTAINER_THREAD_POOL_SIZE, "0");
+
+ if (threadPoolSize > 1 && options.getNumThreadsForProcessElement() <= 1) {
+ // In case the user sets the thread pool through samza config instead options,
+ // set the bundle thread pool size based on container thread pool config
+ // this allows Samza auto-sizing to tune the threads
+ LOG.info("Convert threadPoolSize {} to numThreadsForProcessElement", threadPoolSize);
+ // NumThreadsForProcessElement in option is the source of truth
+ options.setNumThreadsForProcessElement(threadPoolSize);
Review Comment:
Synced up offline with @mynameborat and @jia-gao . We will go with this option precedence approach first to unblock jobs onboarding and then follow up with samza autosizing to figure out precedence.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] xinyuiscool commented on a diff in pull request #26314: Clean up bundle config generation for Samza Runner
Posted by "xinyuiscool (via GitHub)" <gi...@apache.org>.
xinyuiscool commented on code in PR #26314:
URL: https://github.com/apache/beam/pull/26314#discussion_r1170470056
##########
runners/samza/src/main/java/org/apache/beam/runners/samza/translation/ConfigBuilder.java:
##########
@@ -131,6 +117,30 @@ public Config build() {
}
}
+ @VisibleForTesting
+ static Map<String, String> createBundleConfig(
+ SamzaPipelineOptions options, Map<String, String> config) {
+ ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
+ builder.put(MAX_CONCURRENCY, String.valueOf(options.getMaxBundleSize()));
+
+ if (options.getMaxBundleSize() > 1) {
+ final int threadPoolSize = ConfigUtils.asJobConfig(config).getThreadPoolSize();
+ LOG.info("Remove threadPoolSize configs when maxBundleSize > 1");
+ builder.put(JOB_CONTAINER_THREAD_POOL_SIZE, "0");
+ builder.put(JOB_AUTOSIZING_CONTAINER_THREAD_POOL_SIZE, "0");
+
+ if (threadPoolSize > 1 && options.getNumThreadsForProcessElement() <= 1) {
+ // In case the user sets the thread pool through samza config instead options,
+ // set the bundle thread pool size based on container thread pool config
+ // this allows Samza auto-sizing to tune the threads
+ LOG.info("Convert threadPoolSize {} to numThreadsForProcessElement", threadPoolSize);
+ // NumThreadsForProcessElement in option is the source of truth
+ options.setNumThreadsForProcessElement(threadPoolSize);
Review Comment:
Yes, autosizing will override the same user config. But this is a pipeline option in code. It'll be very confusing if we set some parameters in code while it won't be taken into effect in the runtime. To me, this is definitely higher in precedence. The other problem is that we need a way for user to override without turning off autosizing. Our autosizing hasn't been tuned for thread pool size with bundle > 1. So we need to have this capability around.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] xinyuiscool commented on pull request #26314: Clean up bundle config generation for Samza Runner
Posted by "xinyuiscool (via GitHub)" <gi...@apache.org>.
xinyuiscool commented on PR #26314:
URL: https://github.com/apache/beam/pull/26314#issuecomment-1513855771
Run Java PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] jia-gao commented on a diff in pull request #26314: Clean up bundle config generation for Samza Runner
Posted by "jia-gao (via GitHub)" <gi...@apache.org>.
jia-gao commented on code in PR #26314:
URL: https://github.com/apache/beam/pull/26314#discussion_r1170561739
##########
runners/samza/src/main/java/org/apache/beam/runners/samza/translation/ConfigBuilder.java:
##########
@@ -131,6 +117,30 @@ public Config build() {
}
}
+ @VisibleForTesting
+ static Map<String, String> createBundleConfig(
+ SamzaPipelineOptions options, Map<String, String> config) {
+ ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
+ builder.put(MAX_CONCURRENCY, String.valueOf(options.getMaxBundleSize()));
+
+ if (options.getMaxBundleSize() > 1) {
+ final int threadPoolSize = ConfigUtils.asJobConfig(config).getThreadPoolSize();
+ LOG.info("Remove threadPoolSize configs when maxBundleSize > 1");
+ builder.put(JOB_CONTAINER_THREAD_POOL_SIZE, "0");
+ builder.put(JOB_AUTOSIZING_CONTAINER_THREAD_POOL_SIZE, "0");
+
+ if (threadPoolSize > 1 && options.getNumThreadsForProcessElement() <= 1) {
+ // In case the user sets the thread pool through samza config instead options,
+ // set the bundle thread pool size based on container thread pool config
+ // this allows Samza auto-sizing to tune the threads
+ LOG.info("Convert threadPoolSize {} to numThreadsForProcessElement", threadPoolSize);
+ // NumThreadsForProcessElement in option is the source of truth
+ options.setNumThreadsForProcessElement(threadPoolSize);
Review Comment:
To clarify, ` job.container.thread.pool.size` won't work if auto-sizing is enabled.
auto sizing today allows user overrides size configs but still leave auto sizing enabled (asc can still resize the job) by:
1. users can configure `job.autosizing.container.thread.pool.size`. It will be honored when the job starts
2. when auto-sizing resizes the job, it will submit the job to jetty proxy with a new value of
`job.autosizing.container.thread.pool.size` included in the submission config. and write it to the coordinator
stream
3. If user redeploys the job again, the size in 1 is still honored. If the user removes the config in 1 and redeploy,
the size in 2 will be honored (reading from coordinator stream)
for beam job options like ThreadsForProcessElement, asc is not able to achieve 2 so I agree that in ConfigBuilder we should honor the `job.autosizing.container.thread.pool.size` over options specified in code to be able to resize the job. Afterall, if `job.autosizing.container.thread.pool.size` is present, if means either asc resizes the job or the job owner wants to override the size but still want to leave auto-sizing on
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] mynameborat commented on a diff in pull request #26314: Clean up bundle config generation for Samza Runner
Posted by "mynameborat (via GitHub)" <gi...@apache.org>.
mynameborat commented on code in PR #26314:
URL: https://github.com/apache/beam/pull/26314#discussion_r1170536592
##########
runners/samza/src/main/java/org/apache/beam/runners/samza/translation/ConfigBuilder.java:
##########
@@ -131,6 +117,30 @@ public Config build() {
}
}
+ @VisibleForTesting
+ static Map<String, String> createBundleConfig(
+ SamzaPipelineOptions options, Map<String, String> config) {
+ ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
+ builder.put(MAX_CONCURRENCY, String.valueOf(options.getMaxBundleSize()));
+
+ if (options.getMaxBundleSize() > 1) {
+ final int threadPoolSize = ConfigUtils.asJobConfig(config).getThreadPoolSize();
+ LOG.info("Remove threadPoolSize configs when maxBundleSize > 1");
+ builder.put(JOB_CONTAINER_THREAD_POOL_SIZE, "0");
+ builder.put(JOB_AUTOSIZING_CONTAINER_THREAD_POOL_SIZE, "0");
+
+ if (threadPoolSize > 1 && options.getNumThreadsForProcessElement() <= 1) {
+ // In case the user sets the thread pool through samza config instead options,
+ // set the bundle thread pool size based on container thread pool config
+ // this allows Samza auto-sizing to tune the threads
+ LOG.info("Convert threadPoolSize {} to numThreadsForProcessElement", threadPoolSize);
+ // NumThreadsForProcessElement in option is the source of truth
+ options.setNumThreadsForProcessElement(threadPoolSize);
Review Comment:
I feel the inconsistency in behavior is as well confusing to the users. If we want to establish user code as the highest precedence, maybe we should not tie autosizing `job.autosizing.container.thread.pool.size` into this.
Consider this example
1. user configures `job.container.thread.pool.size`, it is reflected in the application behavior
2. autosizing is enabled and user configuration no longer is reflected in the application behavior
3. User sets it in code and the change is reflected in application behavior
user configuration << framework configuration override << user code
Also consider autosizing devs/support team not realizing knobs x, y and z follows auto-sizer values as higher precedence on their end vs a, b follows user values as higher precedence.
>The other problem is that we need a way for user to override without turning off autosizing
I think the recommended way for now is to use auto-sizer tool to overrides specific knobs in case of site-up
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org