[GOBBLIN-1910]Refactor dag manager to reduce bulkiness for adhoc REST calls for launch, resume and kill #3776

meethngala · 2023-09-14T21:49:19Z

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

Description

Here are some details about my PR, including screenshots (if applicable):
I am creating this PR as part of refactoring the DagManager to make it less bulky and eventually stateless (after getting rid of the in-memory references). This PR is only handling the cases for adhoc REST client calls for launch, resume and kill/cancel flows. Currently, this PR is only providing a template for the end-to-end functionality and implementing some parts of existing DagManager's responsibility. There will be subsequent PRs to implement functionalities that are missing likewise scheduler triggers via orchestrator, advancing to the next set of nodes in a Dag, cleaning up of Dags etc.

The method signatures, instantiations and return types use widely accepted and default values, but will be restricted once responsibility/functionality of each class is determined.

Tests

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

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"

phet

nice work. this is a massive change and it's complicated.

high-level thoughts:

in the future, if there's a way to break functionality into multiple PRs, that makes it much easier to review smaller ones. here you might have created the structure of the classes and their declarations w/o putting all the logic within them... just get them to type check, but otherwise throw UnsupportedOperationException
as described in the comments, logic needs reworking between the various methods ITO where it should live. I'll wait for that to settle, before I go over specific logic more exhaustively
because there are so many files, look for a way to put subsets into subpackages rather than into the flat orchestration package. maybe orchestration.proc (should also contain DagProcFactory)?
still, contrary to that advice, I don't believe our convention is to put exceptions in an exception sub-package. that said, those too could go into orchestration.proc (or whatever name you choose)

...n-service/src/main/java/org/apache/gobblin/service/modules/orchestration/AdvanceDagProc.java

...in-service/src/main/java/org/apache/gobblin/service/modules/orchestration/LaunchDagTask.java

...src/main/java/org/apache/gobblin/service/modules/orchestration/NewDagManagerBoilerPlate.java

...n-service/src/main/java/org/apache/gobblin/service/modules/orchestration/WorkInProgress.java

...service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java

phet

for now, first half of second review iteration

gobblin-runtime/src/main/java/org/apache/gobblin/scheduler/JobScheduler.java

...in-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagement.java

.../src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementStateStore.java

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProc.java

...n-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcFactory.java

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagTask.java

...in-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagTaskStream.java

phet

for now, finishing part 2 of second review iteration

...src/main/java/org/apache/gobblin/service/modules/orchestration/NewDagManagerBoilerPlate.java

...n-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagTaskVisitor.java

phet · 2023-09-18T23:50:39Z

...in-service/src/main/java/org/apache/gobblin/service/modules/orchestration/LaunchDagTask.java

+  LaunchDagProc host(DagTaskVisitor visitor) throws IOException, InstantiationException, IllegalAccessException {
+
+    return (LaunchDagProc) visitor.meet(this);
+  }


where does the need arise for an impl of this method to return the specific type of the DagProc? the cast should not be necessary. and do keep in mind that DagTaskVisitor<T> is a generic, hence:

@Override <T> T host(DagTaskVisitor<T> visitor) { return visitor.meet(this); }

Will-Lo · 2023-09-19T17:14:43Z

...service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java

@@ -94,6 +103,9 @@ public DagActionStoreChangeMonitor(String topic, Config config, DagActionStore d
    this.flowCatalog = flowCatalog;
    this.orchestrator = orchestrator;
    this.isMultiActiveSchedulerEnabled = isMultiActiveSchedulerEnabled;
+    // instantiating using default ctor; subsequent PR will handle instantiating with multi-args ctor
+    this.dagTaskStream = new DagTaskStream();
+    this.isRefactoredDagManagerEnabled = ConfigUtils.getBoolean(config, ServiceConfigKeys.GOBBLIN_SERVICE_DAG_MANAGER_ENABLED_KEY, false);


I'd rather you rename this variable as isMultiLeaderDagManagerEnabled otherwise the question becomes why didn't we just modify the base classes.

...src/main/java/org/apache/gobblin/service/modules/orchestration/NewDagManagerBoilerPlate.java

...in-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagTaskStream.java

Will-Lo · 2023-09-19T17:29:25Z

...in-service/src/main/java/org/apache/gobblin/service/modules/orchestration/ResumeDagProc.java

+    }
+
+    if (dag == null) {
+      log.error("Dag " + dagIdToResume + " was found in memory but not found in failed dag state store");


Wouldn't this be in the above check rather than here for this scenario? If Dag is null then whatever object was sent is invalid

.../src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementStateStore.java

umustafi · 2023-09-20T22:11:51Z

...in-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagement.java

+   * @param flowName
+   * @param triggerTimeStamp
+   */
+  void launchFlow(String flowGroup, String flowName, long  triggerTimeStamp);


Leaving it as a note for you in case you want to document behavior in the java doc that adhoc flows triggerTimeStamp will be "0" (see code here), while "-1" indicates some error. All valid values should be >= 0 essentially.

umustafi · 2023-09-20T22:13:09Z

...in-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagement.java

+  void launchFlow(String flowGroup, String flowName, long  triggerTimeStamp);
+
+  /**
+   * Currently, it is handling just the resume of a {@link Dag} request via REST client for adhoc flows


For this and other java docs other than LAUNCH, the "for adhoc flows" is not accurate. Rather it's a resume call from REST client, but they could be resuming a scheduled flow.

...in-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagement.java

umustafi · 2023-09-20T22:22:46Z

...in-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagement.java

+   * @param flowGroup
+   * @param flowName
+   * @param flowExecutionId
+   * @param triggerTimeStamp


add the throws exception to be consistent with above

umustafi · 2023-09-20T22:25:21Z

.../src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementStateStore.java

+  public void addJobState(String dagId, Dag.DagNode<JobExecutionPlan> dagNode);
+
+  public void deleteJobState(String dagId, Dag.DagNode<JobExecutionPlan> dagNode);


when are these actions called? do you ever want to updateJobState?

these will be called when we complete a job in a Dag and based on the job status, we would update it and decide on what should happen using the onJobFinish() method. The impl. for the same will be provided as part of another PR where I add the logic for AdvanceDagProc

...src/main/java/org/apache/gobblin/service/modules/orchestration/processor/CleanUpDagProc.java

...ervice/src/main/java/org/apache/gobblin/service/modules/orchestration/processor/DagProc.java

...ce/src/main/java/org/apache/gobblin/service/modules/orchestration/processor/KillDagProc.java

umustafi · 2023-09-20T23:02:12Z

...lin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/DagTask.java

+   * It is invoked after {@link DagProc#process(DagManagementStateStore)} is completed successfully.
+   * @param multiActiveLeaseArbiter
+   * @throws IOException
+   */


This is different from our approach to launch in that we finish the action before completing lease. Can you document the pros and cons of this method? For ex pro we don't have to do lease arbitration twice for attempting the action and doing it. However now the lease validity time needs to be much larger and include the time to contact executor and carry out the action. Let's note these details here or somewhere else appropriate.

I think there's some confusion here. The lease acquired for a DagTask happens after we have received or processed a message from the CDC stream based on what action needs to be taken on the Dag. Now, we acquire the lease... do our processing on the Dag and upon completion release the lease ... marking it as successful. I agree with the part that lease validity time needs to be increased

Ah you're right I am getting this multi-active part confused with phase 2. Let's add some comments or javadoc to denote potential failures and/or steps needed to complete this event ie: contact executor over network and response time

phet

wow, MASSIVE PR... which is looking better and better--great job, meeth!

phet · 2023-09-21T04:26:04Z

.../src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementStateStore.java

+  public Map<String, LinkedList<Dag.DagNode<JobExecutionPlan>>> getDagToJobs();
+
+  public Map<String, Dag<JobExecutionPlan>> getDagIdToDags();
+
+  public Map<String, Long> getDagToSLA();


for perf, I doubt we'd want to return such massive collections. is it truly necessary to have all the mappings at once? couldn't we instead return only the one related to a particular dag ID?
e.g.:

Dag<JobExecutionPlan> getDags(String dagId); Long getSLA(String dagId);

I agree and it totally makes sense. I have added these additional methods to our DagManagementStateStore interface.

phet · 2023-09-21T04:29:49Z

...-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagerUtils.java

@@ -108,7 +108,7 @@ static DagManager.DagId generateDagId(String flowGroup, String flowName, long fl
    return generateDagId(flowGroup, flowName, String.valueOf(flowExecutionId));
  }

-  static DagManager.DagId generateDagId(String flowGroup, String flowName, String flowExecutionId) {
+  public static DagManager.DagId generateDagId(String flowGroup, String flowName, String flowExecutionId) {


NBD, but I don't find a compelling reason for this, as:

new DagManager.DagId(x, y, z)

is not much different than:

generateDagId(x, y, z);

agreed... only difference being one takes String and the other takes long value for flowExecutionId. These are existing methods defined inside DagManagerUtils... not sure if you want me to define just one and handle the case

I see... if already widely used, no need to rip out a method you did not add. but if only once or twice, I suggest getting rid of it.

contrary to your reply, I do not observe this impl converting between String and long

...n-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcFactory.java

phet · 2023-09-21T04:36:31Z

...vice/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngine.java

+@Slf4j
+public class DagProcessingEngine {
+
+  public static final String DAG_PROCESSING_ENGINE_PREFIX = "gobblin.service.dagProcessingEngine.";


doesn't this need to align w/ what's in ServiceConfigKeys? if so, let's define the prefix over there and bring it in here.

phet · 2023-09-21T04:37:53Z

...vice/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngine.java

+  private DagTaskStream dagTaskStream;
+  private DagProcFactory dagProcFactory;
+  private DagManagementStateStore dagManagementStateStore;
+  private ScheduledExecutorService scheduledExecutorPool;
+  private Config config;
+  private Integer numThreads;
+  private Integer pollingInterval;
+  private DagProcessingEngine.Thread [] threads;
+  private MultiActiveLeaseArbiter multiActiveLeaseArbiter;


many of these seem good candidates for final. what's your take?

phet · 2023-09-21T05:41:15Z

...lin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/DagTask.java

+ */
+
+@Alpha
+public abstract class DagTask<T> {


let's make host a generic method rather than DagTask a generic class

phet · 2023-09-21T05:49:56Z

...n-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcFactory.java

+
+  @Override
+  public DagProc meet(ResumeDagTask resumeDagTask) {
+    throw new UnsupportedOperationException("Currently cannot provide resume proc");


although ResumeDagProc might still throw UOE when process() is invoked... isn't enough scaffolding in place yet to write this as:

return new ResumeDagProc(resumeDagTask);

?

phet · 2023-09-21T05:53:42Z

...ce/src/main/java/org/apache/gobblin/service/modules/orchestration/processor/KillDagProc.java

+public final class KillDagProc extends DagProc {
+
+  private KillDagTask killDagTask;
+  private DagManagementStateStore dagManagementStateStore;


the process method already takes this as a param... do we need to store one in addition to what will be passed to us?

phet · 2023-09-21T05:54:34Z

...ce/src/main/java/org/apache/gobblin/service/modules/orchestration/processor/KillDagProc.java

+  private MetricContext metricContext;
+  private Optional<EventSubmitter> eventSubmitter;


perhaps should these two also be args to process?

phet · 2023-09-21T05:56:03Z

...ce/src/main/java/org/apache/gobblin/service/modules/orchestration/processor/KillDagProc.java

+ */
+@Slf4j
+@Alpha
+public final class KillDagProc extends DagProc {


recall that it's DagProc<S, R>, decide what those types are for kill and then use them in the various method signatures (that currently degraded to Object)

…for launch, resume and kill

…deadline enforcement

phet

finishing for now... part 1 (of 2)

phet · 2023-10-02T17:09:11Z

...in-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagement.java

+ * Responsible for defining the behavior of {@link DagTask} handling scenarios for launch, resume, kill, job start
+ * and flow completion deadlines


rather than "responsible for defining the behavior..." seems more the means by which such tasks are triggered.

that said, may be too much impl. detail to mention DagTask in the javadoc, since I don't see it actually used in the code here

phet · 2023-10-02T17:14:56Z

...in-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagement.java

+   * The eventTimestamp for adhoc flows will always be 0 (zero), while -1 would indicate failures.
+   * Essentially for a valid launch flow, the eventTimestamp needs to be >= 0.
+   * Future implementations will cover launch of flows through the scheduler too!


rather than documenting this to be merely informative, could we instead rephrase as an interface req/policy?

(overall, I'm uncertain on your overall plan for the impl's validation, so this is based solely on reading the javadoc in isolation...)

e.g. throws IllegalArgException when eventTimestamp < 0

Agree with Kip's suggestion here to make it read like an interface
"Handles launching of ... via ..."
@param followed by requirements
@throws if input doesn't match

phet · 2023-10-02T17:17:25Z

.../src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementStateStore.java

+
+
+/**
+ * An interface to provide abstractions for managing {@link Dag} and {@link org.apache.gobblin.service.modules.flowgraph.Dag.DagNode} states


nit: strike "An interface to provide"

add more description like for a Dag, allows one to update or extract its job state or ... giving some high level descriptions

phet · 2023-10-02T17:19:49Z

.../src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementStateStore.java

+  public void addDagSLA(String dagId, Long flowSla);
+
+  public Long getDagSLA(String dagId);


don't we have two kinds? also, since this has never been an SLA, but rather a deadline, would now be an opportunity to adopt the accurate nomenclature?

phet · 2023-10-02T17:21:31Z

.../src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementStateStore.java

+
+  public boolean addFailedDagId(String dagId);
+
+  public boolean checkFailedDagId(String dagId);


unclear: what is meant by "check" - for membership/existence?

phet · 2023-10-02T17:43:05Z

...vice/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngine.java

+            //marks lease success and releases it
+            dagTaskStream.complete(dagTask);


first off, I absolutely LOVE how simple this impl is--even as it segregates responsibility elsewhere for deciding how to process each specific task type.

that said, I wonder whether we might go farther an encapsulate the complete/commit.

e.g. the stream could construct each task w/ a:

@FunctionalInterface interface Committer { void commit(); }

that would internally store the lease. the DagProcFactory would then take that from the task and preserve it in the proc, so the last step of the proc's process could call it

phet · 2023-10-02T17:46:16Z

...n-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagTaskVisitor.java

+ * @param <DagProc>
+ */
+@Alpha
+public interface DagTaskVisitor<DagProc> {


nit: convention would be T (or at the most DagProcType). naming it after an actual type (even though not imported) is quite confusing

phet · 2023-10-02T17:59:20Z

...lin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/DagProc.java

+  abstract protected R act(S state, DagManagementStateStore dagManagementStateStore) throws MaybeRetryableException, Exception;
+  abstract protected void sendNotification(R result, EventSubmitter eventSubmitter) throws MaybeRetryableException, IOException;
+
+  public final void process(DagManagementStateStore dagManagementStateStore, EventSubmitter eventSubmitter, int maxRetryCount, long delayRetryMillis) {


eventually, retries may not be governed by a static count, but rather tracking to how much time remains in the lease

phet · 2023-10-02T18:00:37Z

...lin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/DagProc.java

+      this.sendNotificationWithRetries(result, eventSubmitter, maxRetryCount, delayRetryMillis);
+      log.info("Successfully processed Dag Request");
+    } catch (Exception ex) {
+      throw new RuntimeException("Cannot process Dag Request: ", ex);


I use checked exceptions sparingly, but here, one may actually make sense (rather than RuntimeException)

phet · 2023-10-02T18:17:08Z

...service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/KillDagTask.java

+  public KillDagTask(DagActionStore.DagAction killAction, MultiActiveLeaseArbiter.LeaseObtainedStatus leaseObtainedStatus) {
+
+    this.killAction = killAction;
+    this.leaseObtainedStatusStatus = leaseObtainedStatus;


should this be initialized by a call to super()?

phet

pausing again... part 2

phet · 2023-10-02T20:07:54Z

...service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java

          this.resumesInvoked.mark();
        } else if (dagActionType.equals(DagActionStore.FlowActionType.KILL)) {
          log.info("Received insert dag action and about to send kill flow request");
          dagManager.handleKillFlowRequest(flowGroup, flowName, Long.parseLong(flowExecutionId));
+
+          if(isMultiLeaderDagManagerEnabled) {
+            DagActionStore.DagAction killAction = new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId, DagActionStore.FlowActionType.KILL);


is killFlow called in any other place? if we don't actually have a DagAction already on hand in any of the callsites, not sure it makes sense to use that in the API. perhaps it should be:

killFlow(flowGroup, flowName, flowExecutionId, produceTimestamp)

phet · 2023-10-02T20:09:51Z

...service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java

@@ -80,6 +83,8 @@ public String load(String key) throws Exception {
  protected Orchestrator orchestrator;
  protected boolean isMultiActiveSchedulerEnabled;
  protected FlowCatalog flowCatalog;
+  private DagTaskStream dagTaskStream;


this change monitor shouldn't use a task stream as such (that's exclusively for the DagProcessingEngine.) instead let it write to the DagManagement API

phet · 2023-10-02T20:11:05Z

...service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java

@@ -94,6 +99,9 @@ public DagActionStoreChangeMonitor(String topic, Config config, DagActionStore d
    this.flowCatalog = flowCatalog;
    this.orchestrator = orchestrator;
    this.isMultiActiveSchedulerEnabled = isMultiActiveSchedulerEnabled;
+    // instantiating using default ctor; subsequent PR will handle instantiating with multi-args ctor
+//    this.dagTaskStream = new DagTaskStream();


so this isn't initialized anywhere? won't we get an NPE when using it?

yea we should use guice to bring in it in, see the DagActionStoreChangeMonitorFactory

phet · 2023-10-02T20:14:46Z

...lin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/DagTask.java

+
+
+/**
+ * Defines an individual task or job in a Dag.


"job" already has other connotations as what we run on executors. maybe "defines a singular task in the lifecycle of a managed Dag"?

phet · 2023-10-02T20:15:39Z

...lin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/DagTask.java

+@Alpha
+public abstract class DagTask {
+
+  protected MultiActiveLeaseArbiter.LeaseObtainedStatus leaseObtainedStatusStatus;


why not private final?

phet · 2023-10-02T20:47:34Z

...service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/KillDagProc.java

+  private KillDagTask killDagTask;
+
+  public KillDagProc(KillDagTask killDagTask) {
+    this.killDagTask = killDagTask;
+  }


tip: mark this final and implement via @RequiredArgsConstructor

phet · 2023-10-02T20:49:40Z

...service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/KillDagProc.java

+
+  @Override
+  protected List<Dag.DagNode<JobExecutionPlan>> initialize(DagManagementStateStore dagManagementStateStore) throws IOException {
+    String dagToCancel = this.killDagTask.getKillDagId().toString();


nit: since it's a String, not a dag, I'd name dagId or dagIdToCancel

anyway, for such a short impl, I'd just make a one-liner w/o an intermediate var (unless you're planning to log it)

phet · 2023-10-02T20:50:32Z

...service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/KillDagProc.java

+   * and cancel the job on the executor. The return type is kept as {@link Object} since we might want to refactor
+   * or add more responsibility as part of the actions taken. Hence, after completing all possible scenarios,
+   * it will make sense to update the method signature with its appropriate type.


out of date

phet · 2023-10-02T20:53:54Z

...service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/KillDagProc.java

+    String dagToCancel = this.killDagTask.getKillDagId().toString();
+
+    log.info("Found {} DagNodes to cancel.", dagNodesToCancel.size());
+    for (Dag.DagNode<JobExecutionPlan> dagNodeToCancel : dagNodesToCancel) {
+      killDagNode(dagNodeToCancel);
+    }
+    dagManagementStateStore.getDag(dagToCancel).setFlowEvent(TimingEvent.FlowTimings.FLOW_CANCELLED);
+    dagManagementStateStore.getDag(dagToCancel).setMessage("Flow killed by request");
+    dagManagementStateStore.removeDagActionFromStore(this.killDagTask.getKillDagId(), DagActionStore.FlowActionType.KILL);


nits:

move dagToCancel init after the for loop (and as observed on the name already, it's a dag ID, not a DAG)

call dagMgmtStateStore.getDag(dagToCancel) only once

decide whether still the need for an intermediate dagToCancel var

phet · 2023-10-02T20:54:48Z

...service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/KillDagProc.java

+
+  @Override
+  protected void sendNotification(Dag<JobExecutionPlan> dag, EventSubmitter eventSubmitter) throws MaybeRetryableException {
+    for(Dag.DagNode<JobExecutionPlan> dagNodeToCancel : dag.getNodes()) {


please update your IDE to catch when you forget the space between for (... this happens all over this PR

umustafi · 2023-10-02T21:40:30Z

...rvice/gobblin-rest-api/src/main/snapshot/org.apache.gobblin.rest.jobExecutions.snapshot.json

Is this meant to be a part of ur change? was this done or included by mistake?

umustafi · 2023-10-02T21:42:20Z

...service/src/main/java/org/apache/gobblin/service/modules/core/GobblinServiceGuiceModule.java

+        binder.bind(DagManagementStateStore.class).to(InMemoryDagManagementStateStore.class);
+        binder.bind(DagProcFactory.class).in(Singleton.class);
+        binder.bind(DagProcessingEngine.class).in(Singleton.class);
+        binder.bind(DagTaskStream.class).in(Singleton.class);


I believe you need to create an OptionalBinder like I've done above OptionalBinder.newOptionalBinder(binder,MultiActiveLeaseArbiter.class); as each of these classes are instantiated only in certain cases.

umustafi · 2023-10-02T21:45:06Z

...in-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagement.java

+   * The eventTimestamp for adhoc flows will always be 0 (zero), while -1 would indicate failures.
+   * Essentially for a valid launch flow, the eventTimestamp needs to be >= 0.
+   * Future implementations will cover launch of flows through the scheduler too!


Agree with Kip's suggestion here to make it read like an interface
"Handles launching of ... via ..."
@param followed by requirements
@throws if input doesn't match

umustafi · 2023-10-02T21:47:19Z

.../src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementStateStore.java

+
+
+/**
+ * An interface to provide abstractions for managing {@link Dag} and {@link org.apache.gobblin.service.modules.flowgraph.Dag.DagNode} states


add more description like for a Dag, allows one to update or extract its job state or ... giving some high level descriptions

umustafi · 2023-10-02T21:52:32Z

.../src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementStateStore.java

+  public void addDagSLA(String dagId, Long flowSla);
+
+  public Long getDagSLA(String dagId);


are these dagStart or dagKill Deadlines? We have both configurable so may want to create separate functions for each.

umustafi · 2023-10-02T22:43:10Z

...vice/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/CleanUpDagProc.java

+
+/**
+ * An implementation of {@link DagProc} that is responsible for cleaning up {@link Dag} that has reached an end state
+ * likewise: FAILED, COMPLETE or CANCELED


nit: ie: or including:

umustafi · 2023-10-02T22:44:03Z

...vice/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/CleanUpDagProc.java

+  @Override
+  protected Object initialize(DagManagementStateStore dagManagementStateStore) throws MaybeRetryableException, IOException {
+    throw new UnsupportedOperationException("Not supported");
+


extra space

umustafi · 2023-10-02T22:54:33Z

...lin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/DagProc.java

+    throw new RuntimeException("Max retry attempts reached. Cannot initialize Dag");
+  }
+
+  protected final R actWithRetries(S state, DagManagementStateStore dagManagementStateStore, int maxRetryCount, long delayRetryMillis) {


should we have a generic method that deals with retries and takes a function as parameter? Seems like you are repeating code between these two functions and want to handle them the same way

umustafi · 2023-10-02T23:06:30Z

...service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java

@@ -94,6 +99,9 @@ public DagActionStoreChangeMonitor(String topic, Config config, DagActionStore d
    this.flowCatalog = flowCatalog;
    this.orchestrator = orchestrator;
    this.isMultiActiveSchedulerEnabled = isMultiActiveSchedulerEnabled;
+    // instantiating using default ctor; subsequent PR will handle instantiating with multi-args ctor
+//    this.dagTaskStream = new DagTaskStream();


yea we should use guice to bring in it in, see the DagActionStoreChangeMonitorFactory

umustafi · 2023-10-02T23:08:47Z

...service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java

      if (operation.equals("INSERT")) {
        if (dagActionType.equals(DagActionStore.FlowActionType.RESUME)) {
          log.info("Received insert dag action and about to send resume flow request");
          dagManager.handleResumeFlowRequest(flowGroup, flowName,Long.parseLong(flowExecutionId));
+          //TODO: add a flag for if condition only if multi-active is enabled
          this.resumesInvoked.mark();
        } else if (dagActionType.equals(DagActionStore.FlowActionType.KILL)) {
          log.info("Received insert dag action and about to send kill flow request");
          dagManager.handleKillFlowRequest(flowGroup, flowName, Long.parseLong(flowExecutionId));


the conditional above should be checked first and this should be the else right?

or are we trying a "dummy kill in the new refactor" if so let's add a comment that the change below doesn't carry out the action. it's still in testing.

phet

whew... all done w/ this review cycle

phet · 2023-10-02T21:10:31Z

...in-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagTaskStream.java

+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import org.jetbrains.annotations.NotNull;


looks like the wrong one... are you certain?

phet · 2023-10-03T02:12:41Z

...in-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagTaskStream.java

+  @Getter
+  private final BlockingQueue<DagActionStore.DagAction> dagActionQueue = new LinkedBlockingQueue<>();


a. why a public getter? could it not be solved via a more-targeted method/capability?

b. why always initialize w/ the empty queue, when there's also @AllArgsConstructor?

phet · 2023-10-03T02:16:43Z

...in-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagTaskStream.java

+            if(leaseAttemptStatus instanceof MultiActiveLeaseArbiter.LeaseObtainedStatus) {
+              dagTask = createDagTask(dagAction,
+                  (MultiActiveLeaseArbiter.LeaseObtainedStatus) leaseAttemptStatus);
+            }
+            if (dagTask != null) {
+              break; // Exit the loop when dagTask is non-null
+            }
+          } catch (IOException e) {
+            //TODO: need to handle exceptions gracefully
+            throw new RuntimeException(e);
+          }
+        }
+        return dagTask;


return createDagTask(...)
would be more natural control flow, than is the DagTask dagTask = null;, the break, and finally return dagTask

phet · 2023-10-03T02:17:55Z

...in-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagTaskStream.java

+  private boolean add(DagActionStore.DagAction dagAction) {
+    return this.dagActionQueue.offer(dagAction);
+  }
+
+  private DagActionStore.DagAction take() {
+    return this.dagActionQueue.poll();
+  }


rearrange so public methods come prior to these and other private ones

phet · 2023-10-03T02:21:23Z

...in-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagTaskStream.java

+   */
+
+  @Override
+  public boolean enforceFlowCompletionDeadline(Dag.DagNode<JobExecutionPlan> node) throws ExecutionException, InterruptedException {


still wondering here... do we presume a timer/reminder has been set elsewhere and just triggered?

it seems this simply evaluates whether or not the deadline is exceeded, but doesn't actually take action to enforce. is that true? if so, whose responsibility is the enforcement?

phet · 2023-10-03T02:22:30Z

...in-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagTaskStream.java

+   * Retrieve the {@link JobStatus} from the {@link JobExecutionPlan}.
+   */
+
+  protected JobStatus retrieveJobStatus(Dag.DagNode<JobExecutionPlan> dagNode) {


ping here.... still not clear the extent of the "state" management/retrieval going on (and whether appropriate to this class), since getStatus throws UnsupportedOperationException

phet · 2023-10-03T02:24:59Z

...n/java/org/apache/gobblin/service/modules/orchestration/InMemoryDagManagementStateStore.java

+ * Going forward, each of these in-memory references will be read/write from MySQL store.
+ * Thus, the {@link DagManager} would then be stateless and operate independently.
+ */
+@Getter(onMethod_={@Synchronized})


wondering, since I usually use at field-level, not class level: will it create an accessor for each of the individual constituent collections? I didn't perceive the DagManagementStateStore interface as meant to be quite so low-level... we don't want that do we?

phet · 2023-10-03T02:26:05Z

...n/java/org/apache/gobblin/service/modules/orchestration/InMemoryDagManagementStateStore.java

+  final Map<String, LinkedList<Dag.DagNode<JobExecutionPlan>>> dagToJobs = new HashMap<>();
+  final Map<String, Long> dagToSLA = new HashMap<>();
+  private final Set<String> dagIdstoClean = new HashSet<>();
+  private Optional<DagActionStore> dagActionStore;


why/when would this be empty?

phet reviewed Sep 15, 2023

View reviewed changes

meethngala force-pushed the refactor-dag-manager-to-reduce-bulkiness branch from 88e1596 to cffd894 Compare September 18, 2023 17:26

phet reviewed Sep 18, 2023

View reviewed changes

Will-Lo requested changes Sep 19, 2023

View reviewed changes

umustafi reviewed Sep 20, 2023

View reviewed changes

phet reviewed Sep 21, 2023

View reviewed changes

Meeth Gala added 7 commits September 26, 2023 13:51

initial commit with high-level design for refactoring DagManager

b80ecd1

initial commit for refactoring dag manager handling adhoc REST calls …

ab16ac2

…for launch, resume and kill

address PR comments and add advance and clean up dag tasks

3c15b90

modify to provide impl. for kill dag action type only

262998b

address PR comments and add retries for processing Dag via DagProc

d3000ae

add apis to dag management state store and update logic for flow/job …

74442ea

…deadline enforcement

add back restspec and snapshot json

e675e50

meethngala force-pushed the refactor-dag-manager-to-reduce-bulkiness branch from f53e6b3 to e675e50 Compare September 27, 2023 20:04

phet reviewed Oct 2, 2023

View reviewed changes

umustafi suggested changes Oct 2, 2023

View reviewed changes

phet reviewed Oct 3, 2023

View reviewed changes

		public void addJobState(String dagId, Dag.DagNode<JobExecutionPlan> dagNode);

		public void deleteJobState(String dagId, Dag.DagNode<JobExecutionPlan> dagNode);

		private MetricContext metricContext;
		private Optional<EventSubmitter> eventSubmitter;

		* Responsible for defining the behavior of {@link DagTask} handling scenarios for launch, resume, kill, job start
		* and flow completion deadlines



		/**
		* An interface to provide abstractions for managing {@link Dag} and {@link org.apache.gobblin.service.modules.flowgraph.Dag.DagNode} states

		public void addDagSLA(String dagId, Long flowSla);

		public Long getDagSLA(String dagId);


		public boolean addFailedDagId(String dagId);

		public boolean checkFailedDagId(String dagId);

		//marks lease success and releases it
		dagTaskStream.complete(dagTask);

		@Getter
		private final BlockingQueue<DagActionStore.DagAction> dagActionQueue = new LinkedBlockingQueue<>();

[GOBBLIN-1910]Refactor dag manager to reduce bulkiness for adhoc REST calls for launch, resume and kill #3776

Are you sure you want to change the base?

[GOBBLIN-1910]Refactor dag manager to reduce bulkiness for adhoc REST calls for launch, resume and kill #3776

Conversation

meethngala commented Sep 14, 2023 • edited Loading

JIRA

Description

Tests

Commits

phet left a comment • edited Loading

Choose a reason for hiding this comment

phet left a comment

Choose a reason for hiding this comment

phet left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

phet left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

phet Sep 21, 2023 • edited Loading

Choose a reason for hiding this comment

phet left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

phet left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

phet Oct 2, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

phet Oct 2, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

meethngala commented Sep 14, 2023 •

edited

Loading

phet left a comment •

edited

Loading

phet left a comment •

edited

Loading

phet Sep 21, 2023 •

edited

Loading

phet Oct 2, 2023 •

edited

Loading

phet Oct 2, 2023 •

edited

Loading

phet Oct 3, 2023 •

edited

Loading

phet Oct 3, 2023 •

edited

Loading