[GOBBLIN-2174] GoT YarnService Integration with DynamicScaling

[Blazer-007]

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-2174

Description

☑️ Here are some details about my PR, including screenshots (if applicable):

• In [GOBBLIN-2170] Define Gobblin-on-Temporal dynamic ScalingDirectives with parser and FsScalingDirectiveSource #4068 , Functionality for enabling GoT Dynamic Scaling was introduced.
• This PR aims to extend that functionality to GoT YarnService to have dynamically launch new containers at runtime based on ScalingDirectives

Tests

☑️ My PR adds the following unit tests OR does not need testing for this extremely good reason:

• Tested E2E using newly created DummyDynamicScalingYarnServiceManager which uses DummyScalingDirectiveSource which returns set of fixed profiles and their set points
• YarnServiceTest
  • testBaselineWorkerProfileCreatedWithPassedConfigs and testBuildContainerCommand
• DynamicScalingYarnServiceManagerTest

Logs from container

2024-11-29 05:22:33 PST [,,] INFO  [DynamicScalingYarnService STARTING] org.apache.gobblin.temporal.yarn.YarnService  - Requesting initial containers using baselineWorkerProfile
2024-11-29 05:22:33 PST [,,] INFO  [DynamicScalingYarnService STARTING] org.apache.gobblin.temporal.yarn.YarnService  - Requesting 2 containers with resource=<memory:8192, vCores:2> and allocation request id = Optional.of(0)
2024-11-29 05:22:59 PST [,,] INFO  [DynamicScalingExecutor] org.apache.gobblin.temporal.yarn.DynamicScalingYarnService  - Requesting 2 new containers for profile secondProfile having currently 0 containers
2024-11-29 05:22:59 PST [,,] INFO  [DynamicScalingExecutor] org.apache.gobblin.temporal.yarn.YarnService  - Requesting 2 containers with resource=<memory:2048, vCores:2> and allocation request id = Optional.of(1)
2024-11-29 05:22:59 PST [,,] INFO  [DynamicScalingExecutor] org.apache.gobblin.temporal.yarn.DynamicScalingYarnService  - Requesting 3 new containers for profile firstProfile having currently 0 containers
2024-11-29 05:22:59 PST [,,] INFO  [DynamicScalingExecutor] org.apache.gobblin.temporal.yarn.YarnService  - Requesting 3 containers with resource=<memory:2048, vCores:2> and allocation request id = Optional.of(2)
...
...
2024-11-29 05:23:59 PST [,,] INFO  [DynamicScalingExecutor] org.apache.gobblin.temporal.yarn.DynamicScalingYarnService  - Requesting 1 new containers for profile secondProfile having currently 2 containers
2024-11-29 05:23:59 PST [,,] INFO  [DynamicScalingExecutor] org.apache.gobblin.temporal.yarn.YarnService  - Requesting 1 containers with resource=<memory:2048, vCores:2> and allocation request id = Optional.of(3)
2024-11-29 05:23:59 PST [,,] INFO  [DynamicScalingExecutor] org.apache.gobblin.temporal.yarn.DynamicScalingYarnService  - Requesting 2 new containers for profile firstProfile having currently 3 containers
2024-11-29 05:23:59 PST [,,] INFO  [DynamicScalingExecutor] org.apache.gobblin.temporal.yarn.YarnService  - Requesting 2 containers with resource=<memory:2048, vCores:2> and allocation request id = Optional.of(4)

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"

[Blazer-007]

Previous PR review done here - phet#1

Also please ignore initial commit history as those got added while rebasing

[phet]

looks really good. are there any mocks that would help test YarnService?

[phet]

would it be helpful for unit testing if, rather than hard-coding, this class took the ScalingDirectiveSource FQ class name? I see that could be harder based on the ctor params.

As a simpler alternative, make DynamicScalingYarnServiceManger abstract w/ a method

abstract protected ScalingDirectiveSource createScalingDirectiveSource();

and then the concrete FsSourceDynamicScalingYarnServiceManager would hard code the ScalingDirectiveSource class. you could have a different concrete DSYSM using DummyScalingDirectiveSource. one of those such FQ class names would be a param.

... which reminds me.... how is this DSYSM created and initialized at present?

[Blazer-007]

I am using DummyScalingDirectiveSource(); to launch containers at runtime if i run any job to test complete e2e.

... which reminds me.... how is this DSYSM created and initialized at present?

here after starting yarnservice - https://github.com/apache/gobblin/blob/master/gobblin-temporal/src/main/java/org/apache/gobblin/temporal/yarn/GobblinTemporalApplicationMaster.java#L102 we initialize other service classes whose names are passed through config

gobblin/gobblin-yarn/src/main/java/org/apache/gobblin/yarn/GobblinYarnConfigurationKeys.java

Line 78 in e5d897e

public static final String APP_MASTER_SERVICE_CLASSES = GOBBLIN_YARN_PREFIX + "app.master.serviceClasses";

[Blazer-007]

I agree with point of creating abstract class, let me add that in next commit.
Refactoted the code to use AbstractDSYSM and also added unit tests.

[Blazer-007]

are there any mocks that would help test YarnService?

Have added unit test for buildContainerCommand