[GOBBLIN-2245] Independent Dynamic Scaling for different Activities in Temporal WorkFlow by agam-99 · Pull Request #4159 · apache/gobblin

agam-99 · 2025-12-15T18:14:32Z

Dear Gobblin maintainers,

Please accept this PR. I understand that it will not be reviewed until I have checked off all the steps below!

JIRA

My PR addresses the following Gobblin JIRA issues and references them in the PR title. For example, "[GOBBLIN-XXX] My Gobblin PR"
- https://issues.apache.org/jira/browse/GOBBLIN-2245

Description

This PR introduces a separate memory configuration and dedicated task queue for the work execution phase in Temporal workflows to support better scalability and resource utilization.

Key Changes:

Created dedicated execution task queue (EXECUTION_TASK_QUEUE) to isolate work execution activities from work discovery and commit phases
Added ExecutionWorker to handle execution-stage activities independently
Introduced WorkflowStage enum for stage-specific task queue routing
When dynamic scaling is enabled, NestingExecWorkflow (work execution) routes to the execution queue while other stages use the default queue

Configuration:

GobblinTemporalConfigurationKeys.EXECUTION_TASK_QUEUE - Execution task queue name (default: "GobblinTemporalExecutionTaskQueue")
GobblinTemporalConfigurationKeys.EXECUTION_WORKER_MEMORY_MBS - Memory allocation for execution workers

Benefits:

Independent scaling for work execution vs. discovery/commit phases
More efficient compute utilization

Tests

My PR adds relevant unit tests for the changes made.
All tests use reflection and mocking to test actual method behavior without requiring full Temporal infrastructure.

Commits

My commits all reference JIRA issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "How to write a good git commit message":
1. Subject is separated from body by a blank line
2. Subject is limited to 50 characters
3. Subject does not end with a period
4. Subject uses the imperative mood ("add", not "adding")
5. Body wraps at 72 characters
6. Body explains "what" and "why", not "how"

- unit tests

codecov-commenter · 2025-12-16T06:45:23Z

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 51.71%. Comparing base (6619cb1) to head (e833981).
⚠️ Report is 1 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #4159      +/-   ##
============================================
+ Coverage     49.12%   51.71%   +2.59%     
+ Complexity    10253     7751    -2502     
============================================
  Files          1924     1411     -513     
  Lines         75350    53504   -21846     
  Branches       8361     5882    -2479     
============================================
- Hits          37012    27670    -9342     
+ Misses        35039    23492   -11547     
+ Partials       3299     2342     -957

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

Copilot

Pull request overview

This PR introduces independent dynamic scaling for different workflow stages in Gobblin's Temporal integration by enabling separate memory configuration and dedicated task queues for work execution activities.

Key Changes:

Created dedicated execution task queue and ExecutionWorker class to isolate work execution from discovery/commit phases
Added WorkflowStage enum for stage-specific task queue routing with configurable memory allocation
Modified ProcessWorkUnitsWorkflowImpl to route NestingExecWorkflow to execution queue when dynamic scaling is enabled

Reviewed changes

Copilot reviewed 13 out of 14 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
`GobblinTemporalConfigurationKeys.java`	Added execution worker class constant, execution task queue configuration keys, and stage-specific memory configuration
`ExecutionWorker.java`	New specialized worker for work execution stage with dedicated task queue and concurrency settings
`WorkflowStage.java`	New enum defining workflow stages (WORK_DISCOVERY, WORK_EXECUTION, COMMIT) with stage-specific task queue routing
`AbstractRecommendScalingForWorkUnitsImpl.java`	Enhanced to create profile overlays with ExecutionWorker class and optional memory configuration for scaled containers
`ProcessWorkUnitsWorkflowImpl.java`	Modified to route child workflows to appropriate task queues based on dynamic scaling configuration
`GobblinTemporalTaskRunner.java`	Added initialization logic to start ExecutionWorker in initial container when dynamic scaling is enabled
`AbstractTemporalWorker.java`	Refactored to make config field protected and extracted getTaskQueue() method for subclass customization
`ActivityType.java`	Added overloaded buildActivityOptions() method accepting taskQueue parameter
Test files	Comprehensive unit tests for workflow routing, worker configuration, and scaling behavior
`application.conf`	Minor whitespace cleanup

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2025-12-16T11:08:53Z

.../org/apache/gobblin/temporal/ddm/activity/impl/AbstractRecommendScalingForWorkUnitsImpl.java

+        GobblinTemporalConfigurationKeys.EXECUTION_WORKER_CLASS
+    ));
+
+    return overlayPairs.isEmpty() ? ProfileOverlay.unchanged() : new ProfileOverlay.Adding(overlayPairs);


The check for overlayPairs.isEmpty() on line 100 will always be false because the ExecutionWorker class is unconditionally added to overlayPairs on line 95. This means ProfileOverlay.unchanged() will never be returned. Consider removing this check or restructuring the logic since overlayPairs will always contain at least one element.

Suggested change

return overlayPairs.isEmpty() ? ProfileOverlay.unchanged() : new ProfileOverlay.Adding(overlayPairs);

return new ProfileOverlay.Adding(overlayPairs);

Copilot · 2025-12-16T11:08:53Z

gobblin-temporal/src/test/java/org/apache/gobblin/temporal/ddm/workflow/WorkflowStageTest.java

+
+    // Verify
+    Assert.assertEquals(taskQueue, GobblinTemporalConfigurationKeys.DEFAULT_GOBBLIN_TEMPORAL_TASK_QUEUE,
+        "WORK_COMMIT should use default task queue");


The comment mentions "WORK_COMMIT" but the actual enum value being tested is "COMMIT". Update the comment to match the enum value for consistency.

Copilot · 2025-12-16T11:08:54Z

gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/activity/ActivityType.java

+  public ActivityOptions buildActivityOptions(Properties props, boolean setHeartbeatTimeout, String taskQueue) {
+    ActivityOptions.Builder builder = ActivityOptions.newBuilder()
+        .setStartToCloseTimeout(getStartToCloseTimeout(props))
+        .setRetryOptions(buildRetryOptions(props))
+        .setTaskQueue(taskQueue);
+
+    if (setHeartbeatTimeout) {
+      builder.setHeartbeatTimeout(getHeartbeatTimeout(props));
+    }
+
+    return builder.build();
+  }
+


The newly added buildActivityOptions method with taskQueue parameter is not used anywhere in the codebase. If this method is intended for future use, consider documenting this in a code comment. Otherwise, consider removing it to avoid maintaining unused code. Alternatively, if this should be used for routing activities to the execution queue when dynamic scaling is enabled, the implementation may be incomplete.

Suggested change

public ActivityOptions buildActivityOptions(Properties props, boolean setHeartbeatTimeout, String taskQueue) {

ActivityOptions.Builder builder = ActivityOptions.newBuilder()

.setStartToCloseTimeout(getStartToCloseTimeout(props))

.setRetryOptions(buildRetryOptions(props))

.setTaskQueue(taskQueue);

if (setHeartbeatTimeout) {

builder.setHeartbeatTimeout(getHeartbeatTimeout(props));

}

return builder.build();

}

Copilot · 2025-12-16T11:08:54Z

gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/worker/ExecutionWorker.java

+ */
+public class ExecutionWorker extends AbstractTemporalWorker {
+    public static final long DEADLOCK_DETECTION_TIMEOUT_SECONDS = 120;
+    public int maxExecutionConcurrency;


The field maxExecutionConcurrency should be declared as private or have documentation explaining why it needs package-private visibility. Consider making it private if external access is not required, or document the reason for package-private visibility if it's intentional for testing purposes.

Suggested change

public int maxExecutionConcurrency;

private int maxExecutionConcurrency;

Copilot · 2025-12-16T11:08:54Z

...lin-temporal/src/main/java/org/apache/gobblin/temporal/GobblinTemporalConfigurationKeys.java

@@ -134,4 +141,5 @@ public interface GobblinTemporalConfigurationKeys {
  String TEMPORAL_ACTIVITY_RETRY_OPTIONS_MAXIMUM_ATTEMPTS = TEMPORAL_ACTIVITY_RETRY_OPTIONS + "maximum.attempts";
  int DEFAULT_TEMPORAL_ACTIVITY_RETRY_OPTIONS_MAXIMUM_ATTEMPTS = 4;



Add a documentation comment for the WORK_EXECUTION_MEMORY_MB configuration key to explain its purpose and usage, similar to other configuration keys in this interface. The comment should clarify that this is the memory allocation in megabytes for execution worker containers when dynamic scaling is enabled.

Suggested change

/**

* Memory allocation (in megabytes) for execution worker containers when dynamic scaling is enabled.

* This value determines the amount of memory assigned to each worker container during execution.

*/

pratapaditya04 · 2026-01-06T06:16:05Z

conf/yarn/application.conf

 # job history store ( WARN [GobblinYarnAppLauncher] NOT starting the admin UI because the job execution info server is NOT enabled )
 job.execinfo.server.enabled=false
-job.history.store.enabled=false
+job.history.store.enabled=false


nit: remove the whitespace change

pratapaditya04 · 2026-01-06T06:18:40Z

gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/worker/ExecutionWorker.java

+
+    public ExecutionWorker(Config config, WorkflowClient workflowClient) {
+        super(config, workflowClient);
+        this.maxExecutionConcurrency = ConfigUtils.getInt(config, GobblinTemporalConfigurationKeys.TEMPORAL_NUM_THREADS_PER_WORKER,


Can this config be process workunit specific instead of using a generic config?

yeah probably we can in another PR

We should add this in same PR, its just adding one more config in GobblinTemporalConfigurationKeys, default value can be same

String TEMPORAL_NUM_THREADS_PER_EXECUTION_WORKER = PREFIX + "num.threads.per.execution.worker";

pratapaditya04 · 2026-01-06T06:25:47Z

...in-temporal/src/main/java/org/apache/gobblin/temporal/cluster/GobblinTemporalTaskRunner.java

+    executionWorker.start();
+    workers.add(executionWorker);
+    logger.info("Worker started for class: {}", GobblinTemporalConfigurationKeys.EXECUTION_WORKER_CLASS);
+  }


Have we tested the scenario with 1 container?

pratapaditya04 · 2026-01-06T06:28:44Z

...rc/main/java/org/apache/gobblin/temporal/ddm/workflow/impl/ProcessWorkUnitsWorkflowImpl.java

+        .setWorkflowId(Help.qualifyNamePerExecWithFlowExecId(COMMIT_STEP_WORKFLOW_ID_BASE, config))
        .build();

+    // CommitStepWorkflow inherits default queue from ProcessWorkUnitsWorkflow parent


Where are we setting default queue for CommitActivity? We don't want commit activity in the process queue right?

Are we saying that we have only set the task queu for the child worfklow/activities of ProcessWorkUnitsWorkflow parent. so the CommitStep would default to the default Task queue?

yes correct

pratapaditya04 · 2026-01-06T06:34:56Z

gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/worker/ExecutionWorker.java

+    protected WorkerOptions createWorkerOptions() {
+        return WorkerOptions.newBuilder()
+            .setDefaultDeadlockDetectionTimeout(TimeUnit.SECONDS.toMillis(DEADLOCK_DETECTION_TIMEOUT_SECONDS))
+            .setMaxConcurrentActivityExecutionSize(this.maxExecutionConcurrency)


Do you think it makes sense to have different configs for this? Currently this is being driven by TEMPORAL_NUM_THREADS_PER_WORKER, but suppose If I want to reduce the number of threads in the worker but don't want to reduce the execution local activity concurrenncy.

yeah good point - I am not sure about this, maybe this can be discussed and handled in another PR.
This PR only handles independent memory configuration and other configs remain same.

Lets add this in same PR too, so that we have configs to play around and tune if needed

Blazer-007 · 2026-01-13T07:26:24Z

gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/worker/ExecutionWorker.java

+
+    public ExecutionWorker(Config config, WorkflowClient workflowClient) {
+        super(config, workflowClient);
+        this.maxExecutionConcurrency = ConfigUtils.getInt(config, GobblinTemporalConfigurationKeys.TEMPORAL_NUM_THREADS_PER_WORKER,


We should add this in same PR, its just adding one more config in GobblinTemporalConfigurationKeys, default value can be same

String TEMPORAL_NUM_THREADS_PER_EXECUTION_WORKER = PREFIX + "num.threads.per.execution.worker";

Blazer-007 · 2026-01-13T07:28:09Z

gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/worker/ExecutionWorker.java

+    protected WorkerOptions createWorkerOptions() {
+        return WorkerOptions.newBuilder()
+            .setDefaultDeadlockDetectionTimeout(TimeUnit.SECONDS.toMillis(DEADLOCK_DETECTION_TIMEOUT_SECONDS))
+            .setMaxConcurrentActivityExecutionSize(this.maxExecutionConcurrency)


Lets add this in same PR too, so that we have configs to play around and tune if needed

Blazer-007 · 2026-01-13T07:28:56Z

gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/worker/ExecutionWorker.java

+    /**
+     * Package-private for testing purposes.
+     */
+    int getMaxExecutionConcurrency() {
+        return maxExecutionConcurrency;
+    }


You can use lombok @Getter(AccessLevel.PROTECTED) for this

Blazer-007 · 2026-01-13T09:10:45Z

gobblin-temporal/src/test/java/org/apache/gobblin/temporal/ddm/worker/ExecutionWorkerTest.java

+   */
+  private ExecutionWorker createMockWorker(Config config) throws Exception {
+    ExecutionWorker worker = Mockito.mock(ExecutionWorker.class, Mockito.CALLS_REAL_METHODS);
+


If we are writing unit test for ExecutionWorker, we shouldn't be mocking that class itself, you can mock Config and WorkflowClient passed into constructor

Blazer-007 · 2026-01-13T09:13:55Z

gobblin-temporal/src/test/java/org/apache/gobblin/temporal/ddm/worker/ExecutionWorkerTest.java

+
+  /**
+   * Helper to invoke the protected getWorkflowImplClasses method using reflection.
+   */


Protected members should be directly accessible if package for Main Class and Test Class are same, can you check this again as I don't think use of reflection is needed

.../org/apache/gobblin/temporal/ddm/activity/impl/AbstractRecommendScalingForWorkUnitsImpl.java

Agam Pal Singh and others added 11 commits December 8, 2025 14:41

independent memory and container configuration for temporal stages

0ca2cb7

route non execute activities to default queue

383a7cd

code refactoring

a428b4b

code cleanup

7b98621

fix: spin up 1 container initially by default

af5e22b

proper queue routing for commit phase

c6f8647

code cleanup

4bdc43a

- spin up initial container with execution worker

c31b0ca

- unit tests

- only forward NestingExecWorkflow to execution queue

bb49427

removed redundant unit tests

85959e3

removed redundant unit tests

82b4dc0

agam-99 force-pushed the feat/dynamic-stage-scaling branch from 1196a45 to 82b4dc0 Compare December 16, 2025 06:15

Merge branch 'master' into feat/dynamic-stage-scaling

04b2b6c

Blazer-007 requested a review from Copilot December 16, 2025 10:53

Copilot started reviewing on behalf of Blazer-007 December 16, 2025 10:53 View session

Copilot AI reviewed Dec 16, 2025

View reviewed changes

copilot review fixes

3528f88

pratapaditya04 reviewed Jan 6, 2026

View reviewed changes

agam-99 added 3 commits January 7, 2026 11:53

Merge branch 'master' into feat/dynamic-stage-scaling

f0852b1

PR review fixes

3fad0e6

undo whitespace change

c05f8d9

Blazer-007 reviewed Jan 13, 2026

View reviewed changes

agam-99 added 3 commits January 14, 2026 10:31

PR review comments addressing

e94fb27

PR review comments addressing

9a550ed

updated property name

9ff14c6

pratapaditya04 approved these changes Jan 15, 2026

View reviewed changes

removed unused import

e833981

Blazer-007 changed the title ~~Independent Dynamic Scaling for different Activities in Temporal WorkFlow~~ [GOBBLIN-2245] Independent Dynamic Scaling for different Activities in Temporal WorkFlow Jan 15, 2026

Blazer-007 approved these changes Jan 15, 2026

View reviewed changes

Blazer-007 merged commit 8e9146e into apache:master Jan 15, 2026
6 checks passed

	return overlayPairs.isEmpty() ? ProfileOverlay.unchanged() : new ProfileOverlay.Adding(overlayPairs);
	return new ProfileOverlay.Adding(overlayPairs);

	public int maxExecutionConcurrency;
	private int maxExecutionConcurrency;

		@@ -134,4 +141,5 @@ public interface GobblinTemporalConfigurationKeys {
		String TEMPORAL_ACTIVITY_RETRY_OPTIONS_MAXIMUM_ATTEMPTS = TEMPORAL_ACTIVITY_RETRY_OPTIONS + "maximum.attempts";
		int DEFAULT_TEMPORAL_ACTIVITY_RETRY_OPTIONS_MAXIMUM_ATTEMPTS = 4;

+  /**
+   * Memory allocation (in megabytes) for execution worker containers when dynamic scaling is enabled.
+   * This value determines the amount of memory assigned to each worker container during execution.
+   */

Conversation

agam-99 commented Dec 15, 2025 • edited by Blazer-007 Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

JIRA

Description

Tests

Commits

Uh oh!

codecov-commenter commented Dec 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Dec 16, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 16, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 16, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 16, 2025

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 16, 2025

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

agam-99 commented Dec 15, 2025 •

edited by Blazer-007

Loading

codecov-commenter commented Dec 16, 2025 •

edited

Loading