[draft] Dag refactor #3825

arjun4084346 · 2023-11-12T04:47: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-XXX

Description

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

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"

codecov-commenter · 2023-11-13T19:16:48Z

Codecov Report

Attention: 470 lines in your changes are missing coverage. Please review.

Comparison is base (e392c97) 47.60% compared to head (a9d3ca0) 47.37%.

Files	Patch %	Lines
...n/service/modules/orchestration/NewDagManager.java	0.00%	167 Missing ⚠️
...orchestration/InMemoryDagManagementStateStore.java	0.00%	46 Missing ⚠️
...ervice/modules/orchestration/proc/KillDagProc.java	0.00%	36 Missing ⚠️
...ice/modules/orchestration/DagProcessingEngine.java	0.00%	26 Missing ⚠️
...gobblin/service/monitoring/JobStatusRetriever.java	0.00%	22 Missing ⚠️
...service/modules/orchestration/DagManagerUtils.java	54.76%	16 Missing and 3 partials ⚠️
...n/service/modules/orchestration/DagTaskStream.java	0.00%	16 Missing ⚠️
.../service/modules/orchestration/DagProcFactory.java	0.00%	15 Missing ⚠️
...in/service/modules/orchestration/proc/DagProc.java	0.00%	14 Missing ⚠️
...in/service/modules/orchestration/task/DagTask.java	0.00%	8 Missing ⚠️
... and 23 more

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #3825      +/-   ##
============================================
- Coverage     47.60%   47.37%   -0.23%     
- Complexity    11066    11073       +7     
============================================
  Files          2159     2181      +22     
  Lines         85471    85938     +467     
  Branches       9497     9523      +26     
============================================
+ Hits          40685    40713      +28     
- Misses        41081    41522     +441     
+ Partials       3705     3703       -2

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

umustafi

my initial thoughts on PR, will go in depth to check everything.

umustafi · 2023-12-14T01:09:44Z

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

+//  public void addDag(Dag<JobExecutionPlan> dag);
+
+  public void addJobState(String dagId, Dag.DagNode<JobExecutionPlan> dagNode);
+
+  public void deleteJobState(String dagId, Dag.DagNode<JobExecutionPlan> dagNode);
+
+  public boolean hasRunningJobs(String dagId);
+
+  public void removeDagActionFromStore(DagActionStore.DagAction dagAction) throws IOException;
+


put TODO for the commented out methods?

umustafi · 2023-12-14T01:12:47Z

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

+          DagActionStore.DagAction dagAction = DagManagerUtils.createDagAction(flowGroup, flowName, String.valueOf(flowExecutionId), DagActionStore.FlowActionType.KILL);
+          this.dagProcessingEngine.get().addDagAction(dagAction);


At this point we should also/instead add a line to store the new dag action in dagAction store so we can receive this in changeMonitor across all hosts. Remember we want all actions persisted in MySQL and then coordination across hosts.

umustafi · 2023-12-14T01:15:58Z

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

+      return;
+    }
+    // todo - find next dag node to run
+    Dag.DagNode<JobExecutionPlan> dagNode = new Dag.DagNode<>(new JobExecutionPlan(JobSpec.builder().build(), null));


We should use dagId to retrieve DagNode from DagManagementStateStore (the next dagNode without a status)

umustafi · 2023-12-14T01:18:33Z

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

+@Slf4j
+public abstract class DagProc<T extends DagTask> {
+  protected final DagProcFactory dagProcFactory;
+  protected final UserQuotaManager quotaManager;


Should checking quota happen in DagProc or in TaskStream (dagProc needs dagNode to check).

can we make the quota checking a util function as we may need to reuse for launch and advance and retry/resume

umustafi · 2023-12-14T01:24:16Z

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

+      //flowTriggerHandler.getLeaseOnDagAction(jobProps, dagAction, System.currentTimeMillis());
+      //if (leaseAttemptStatus instanceof MultiActiveLeaseArbiter.LeaseObtainedStatus) {
+        // can it return null? is this iterator allowed to return null?
+        return createDagTask(dagAction, new MultiActiveLeaseArbiter.LeaseObtainedStatus(dagAction, System.currentTimeMillis()));


For now we are spoofing getting LeaseObtainedStatus since there is no handler at the moment, but we should have a while loop that continues polling until we get a DagTask that we obtain lease for. Let's still encode that logic.

while lease not obtained { // poll dag action queue // attempt lease on it }

Adding that will create findBugsMain error

umustafi · 2023-12-14T01:29:38Z

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

+public class NewDagManager implements DagManagement {
+  public static final String DAG_MANAGER_PREFIX = "gobblin.service.dagManager.";
+
+  private static final Integer DEFAULT_JOB_STATUS_POLLING_INTERVAL = 10;


Is this copy of NewDagManager needed or can existing DagManager be re-used? What are the key differences here? The main change we want to make is that we have an abstraction around obtaining/reading dags through the DagManagementStateStore to interchangeably use in memory versus mysql dag storage right.

umustafi · 2023-12-14T01:31:59Z

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

+  private final Map<String, Dag<JobExecutionPlan>> dags = new HashMap<>();
+  private final Map<String, Dag<JobExecutionPlan>> resumingDags = new HashMap<>();
+  private static final long DAG_FLOW_STATUS_TOLERANCE_TIME_MILLIS = TimeUnit.MINUTES.toMillis(5);
+
+
+  private final Map<Dag.DagNode<JobExecutionPlan>, Dag<JobExecutionPlan>> jobToDag = new HashMap<>();
+  private final Map<String, Dag.DagNode<JobExecutionPlan>> dagNodes = new HashMap<>();
+  private final Map<String, LinkedList<Dag.DagNode<JobExecutionPlan>>> dagToJobs = new HashMap<>();
+  final Map<String, Long> dagToSLA = new HashMap<>();
+  DagManager.DagManagerThread[] dagManagerThreads;


DagManagement has api to use DagManagementStateStore and we should remove these in-memory references

gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatusRetriever.java

umustafi · 2023-12-14T21:56:17Z

gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/Dag.java

+      long flowExecutionId = ConfigUtils.getLong(jobConfig, ConfigurationKeys.FLOW_EXECUTION_ID_KEY, 0L);
+      String jobName = ConfigUtils.getString(jobConfig, ConfigurationKeys.JOB_NAME_KEY, "");
+      String jobGroup = ConfigUtils.getString(jobConfig, ConfigurationKeys.JOB_GROUP_KEY, "");


can we codify the defaults used for jobname/group, execution id if none are provided in ConfigurationKeys? We may need again in other locations.

umustafi · 2023-12-14T22:00:11Z

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

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


can u add brief java doc to explain how these are used? I expect these methods to be defined in DagStateStore and these are a wrapper around those and these methods are meant to be used only internally and should be private/protected.

addJobState is called during dag initialization. It was used privately because DagManager does dag initialization. But now when we are moving stuffs away, DagProc is doing dag initialization and is using it from outside.

umustafi · 2023-12-14T22:00:40Z

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

+  void addJobState(String dagId, Dag.DagNode<JobExecutionPlan> dagNode);
+
+  /**
+   * defines what to do when a job (dag node) finishes


nit: capitalize first word here and in other docs

umustafi · 2023-12-14T22:02:37Z

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

+  Map<String, Set<Dag.DagNode<JobExecutionPlan>>> submitNext(String dagId) throws IOException;
+  void removeDagActionFromStore(DagManager.DagId dagIdToResume, DagActionStore.FlowActionType flowActionType)
+      throws IOException;
+
+  void handleJobStatusEvent(JobStatusEvent jobStatusEvent);
+  void handleKillFlowEvent(KillFlowEvent killFlowEvent);


let's be consistent with spacing, this is a bit hard to read. Can you add sections for the methods? ie:

// High Level flow actions void handleKillFlowEvent...

umustafi · 2023-12-14T22:35:17Z

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

+      //flowTriggerHandler.getLeaseOnDagAction(jobProps, dagAction, System.currentTimeMillis());
+      //if (leaseAttemptStatus instanceof MultiActiveLeaseArbiter.LeaseObtainedStatus) {
+        // can it return null? is this iterator allowed to return null?
+        return createDagTask(dagAction, new MultiActiveLeaseArbiter.LeaseObtainedStatus(dagAction, System.currentTimeMillis()));


while lease not obtained { // poll dag action queue // attempt lease on it }

umustafi · 2023-12-14T22:35:56Z

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

+      case RESUME:
+      case LAUNCH:
+      case ADVANCE:


add TODOs to complete later

umustafi · 2023-12-14T22:36:19Z

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

+    //dagTask.conclude(this.flowTriggerHandler.getMultiActiveLeaseArbiter());
+  }
+
+  public void addDagAction(DagActionStore.DagAction dagAction) {


clarify name "addDagActionToQueue"

umustafi · 2023-12-14T22:43:05Z

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

+      boolean flowKilled = enforceFlowStartDeadline(node);
+      boolean jobKilled = false;
+
+      if (!flowKilled) {
+        JobStatus jobStatus = pollJobStatus(node);
+        try {
+          jobKilled = enforceJobCompletionDeadline(node, jobStatus);


do we try to enforce the SLA kills first even when we receive an API request? I was not aware this is the norm. Is this not supposed to kill the dag because of api request or its used differently?

umustafi · 2023-12-14T22:43:44Z

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

+  private JobStatus pollStatus(String flowGroup, String flowName, long flowExecutionId, String jobGroup, String jobName) {
+    long pollStartTime = System.nanoTime();
+    Iterator<JobStatus> jobStatusIterator =
+        this.dagManager.getJobStatusRetriever().getJobStatusesForFlowExecution(flowName, flowGroup, flowExecutionId, jobName, jobGroup);
+    Instrumented.updateTimer(this.dagManager.getJobStatusPolledTimer(), System.nanoTime() - pollStartTime, TimeUnit.NANOSECONDS);
+
+    if (jobStatusIterator.hasNext()) {
+      return jobStatusIterator.next();
+    } else {
+      return null;
+    }
+  }


this method seems like it would be used by other action threads too. can we put this into a util class?

umustafi · 2023-12-14T23:09:55Z

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

+@Slf4j
+public abstract class DagProc<T extends DagTask> {
+  protected final DagProcFactory dagProcFactory;
+  protected final UserQuotaManager quotaManager;


can we make the quota checking a util function as we may need to reuse for launch and advance and retry/resume

umustafi · 2023-12-14T23:11:26Z

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

+    try {
+      quotaManager.checkQuota(Collections.singleton(dagNode));
+
+      producer = DagManagerUtils.getSpecProducer(dagNode);
+      TimingEvent jobOrchestrationTimer = this.eventSubmitter.isPresent() ? this.eventSubmitter.get().
+          getTimingEvent(TimingEvent.LauncherTimings.JOB_ORCHESTRATED) : null;
+
+      // Increment job count before submitting the job onto the spec producer, in case that throws an exception.
+      // By this point the quota is allocated, so it's imperative to increment as missing would introduce the potential to decrement below zero upon quota release.
+      // Quota release is guaranteed, despite failure, because exception handling within would mark the job FAILED.
+      // When the ensuing kafka message spurs DagManager processing, the quota is released and the counts decremented
+      // Ensure that we do not double increment for flows that are retried
+      if (dagNode.getValue().getCurrentAttempts() == 1) {
+        dagManagerMetrics.incrementRunningJobMetrics(dagNode);
+      }


repeated code here, it seems like we have scope to consolidate

umustafi · 2023-12-14T23:14:32Z

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

+ */
+@Slf4j
+@Alpha
+// todo - maybe reload need different treatment from other operations, because these are already present in the dag store


when is reload used? It will be helpful to describe why we need reload, is this only on startup?

yes this is only on startup

do we still need this? what is loading here? every time we start up we only need to initialize the scheduler then the scheduler should take care of triggering launch tasks. or if there are unfinished tasks in the dagActionStore those will be separately dealt with by the startup sequence.

umustafi · 2023-12-14T23:16:17Z

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

+@Alpha
+public final class KillDagProc extends DagProc<KillDagTask> {
+
+  // should dag task be a part of dag proc?


do we already have access to the KillDagTask if we're extending it? it does make sense to me to have access to the original task associated with the Proc

umustafi · 2023-12-14T23:17:36Z

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

@@ -154,9 +165,15 @@ protected void processMessage(DecodeableKafkaRecord message) {
        log.info("DagAction change ({}) received for flow: {}", dagActionType, dagAction);
        if (dagActionType.equals(DagActionStore.FlowActionType.RESUME)) {
          dagManager.handleResumeFlowRequest(flowGroup, flowName,Long.parseLong(flowExecutionId));
+          if (isMultiLeaderDagManagerEnabled) {


should we remove this for now if we are also not doing launch? do u want to test only kill or kill and resume in first iteration?

umustafi · 2023-12-15T00:25:09Z

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

+
+
+/**
+ * NewDagManager manages dags in memory and various mappings.


The job of the newDagManager is actually to only add new work ("tasks") to the dagTaskStream. Let's update the description.

This should be a lightweight class that is only adding new tasks. Do we want all this logic for status retrieving & clean up after a task finishes? Need to decide who polls for job status after it completes and cleans up or submits a next job

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

umustafi · 2023-12-15T00:39:15Z

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

+    }
+    if (dagActionStore.isPresent()) {
+      Collection<DagActionStore.DagAction> dagActions = dagActionStore.get().getDagActions();
+      for (DagActionStore.DagAction action : dagActions) {


#3841 moved in recent PR if u rebase. all of these are in dagActionStoremonitor and we should call DagManagement interface to handle these

umustafi · 2023-12-15T00:42:50Z

gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/KafkaJobStatusMonitor.java

@@ -138,7 +146,8 @@ public <V> void onRetry(Attempt<V> attempt) {
            }
          }
        }));
-        this.eventProducer = observabilityEventProducer;
+    this.eventProducer = observabilityEventProducer;
+    this.dagProcessingEngine = dagProcessingEngine;


Why are we interacting with dagProcessingEngine? We should be passing work to newDagManager through event bus right? Also can you clarify this in comment that it is being passed to newDagManager

phet

this should be a good start, while I get to the last few files a bit later

gobblin-api/src/main/java/org/apache/gobblin/service/ServiceConfigKeys.java

...gobblin-kafka-09/src/test/java/org/apache/gobblin/runtime/KafkaAvroJobStatusMonitorTest.java

phet · 2023-12-19T11:27:10Z

gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/KafkaJobStatusMonitor.java

@@ -94,6 +100,8 @@ public abstract class KafkaJobStatusMonitor extends HighLevelConsumer<byte[], by
  @Getter
  private final StateStore<org.apache.gobblin.configuration.State> stateStore;
  private final ScheduledExecutorService scheduledExecutorService;
+  @Getter(AccessLevel.PUBLIC)


I'm not sure whether I'm having trouble w/ the diff... but I can't find anyone accessing this.

correspondingly, I can't tell the intention behind it. that really deserves comments to document plus a name to indicate the purpose of this EventBus (and perhaps also the type of its payload), as opposed to merely repeating its type name

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

phet · 2023-12-19T11:38:56Z

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

+      if (isMultiLeaderDagManagerEnabled) {
+        this.dagProcessingEngine.addDagAction(dagAction);
+      }


the Null Pattern is cleaner than every time checking a separate boolean. i.e. define an interface, where one impl has a addDagAction that actually does real processing, and another "no-op" impl where addDagAction does nothing. then instantiate the NoOp if isMultiActive == false

phet · 2023-12-19T12:17:46Z

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

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


I don't see where you're using the type param, T. the specific DagTask derived type can probably be passed into the ctor for the DagProc derived type.

generics would come in handy though for:
a. the return type of initialize() (which is the param type of act())
b. the return type of act() (which is the param type of commit())

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

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

phet · 2023-12-19T12:23:50Z

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

+  public DagProc host(DagTaskVisitor<DagProc> visitor) {
+    return visitor.meet(this);
+  }


doesn't this work:

<T extends DagProc> public T host(DTV<T> visitor) { return visitor.meet(this);

?

remove changes not related to dag manager refactoring commit before preemption use visiter pattern store dag action instead of dag task in dag task stream use dag management store in NewDagManager merge conflicts some cleaning fix tests

umustafi · 2023-12-19T23:24:46Z

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

We can remove this task type as well as Proc because we load up tasks as needed when there is a dagAction pending. Steady state active host (or a newly starting up one) will load dag from dag state store after getting lease for the dagTask (that was added to the dagTaskStream).

umustafi · 2023-12-19T23:25:14Z

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

see comment on ReloadDagTask

umustafi · 2023-12-19T23:31:45Z

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

+    }
+    this.dagStateStore.writeCheckpoint(this.dag);
+    for (Dag.DagNode<JobExecutionPlan> dagNode : this.dag.getStartNodes()) {
+      this.dagProcFactory.dagProcessingEngine.addAdvanceDagAction(dagNode);


overall flow dagAction stored in dagActionStore -> dagActionStoreMonitor receives action and adds task to dagTaskStream-> dagProcessingEngine iterates through tasks and "acts" on each one to do work via dagProc

umustafi · 2023-12-19T23:35:47Z

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

+      return;
+    }
+    this.dagStateStore.writeCheckpoint(this.dag);
+    for (Dag.DagNode<JobExecutionPlan> dagNode : this.dag.getStartNodes()) {


Two options for how to deal with multi-hop flow

the way you have here

update status for dagStateStore

then create separate dagActions for each dagNode

launch 1 dagNode at a time, upon completion check for more nodes to launch and add them one by one committing each time

umustafi · 2023-12-19T23:40:31Z

.../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
+ * and allows add/delete and other functions.


add some detail in java doc: load/update/save any state relating to dag. used by dagProcs

umustafi · 2023-12-19T23:46:15Z

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

+  void removeDagActionFromStore(DagActionStore.DagAction dagAction) throws IOException;
+  void handleJobStatusEvent(JobStatusEvent jobStatusEvent);
+  void handleKillFlowEvent(KillFlowEvent killFlowEvent);
+  void handleResumeFlowEvent(ResumeFlowEvent resumeFlowEvent) throws IOException;


these are the core functionality of dagManagement (handling new dag action, enforce SLA/resume/launch/kill ... and add to stream). the other methods below should be in dagManagementStateStore and dagProc interact with dagManagementStateStore

umustafi · 2023-12-20T00:02:45Z

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

+ * e) load {@link Dag}s on service-start / set-active.
+ */
+@Slf4j
+public class NewDagManager implements DagManagement {


if these methods already exist elsewhere we should remove or see if we can abstract away unnecessary methods and functionality. We can clean up after ironing out interface for everything else. For now let's keep b).
Instead of c) for start/kill SLA enforcement we need to send some action to revisit and check if started/completed but all hosts need to know of it so need to durably store it somewhere (timer). Then only if SLA breached we add a dagAction to kill. launchDagAction -> durably storesetStartSlaTimer AND setCompletionSlaTimer somewhere and set the timer for urself (on restart we can load these in or have another stream for it). When all host timers go off, do lease arbitration to enforce and emit a cancelBCStartSlaBreached/CompletionSlaBreached action

umustafi · 2024-01-03T22:52:54Z

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

+import org.apache.gobblin.service.monitoring.event.JobStatusEvent;
+
+
+public interface DagManagement {


shouldn't we have a method to handle a launch?

umustafi · 2024-01-03T23:00:31Z

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

+  @Override
+  public Object meet(ReloadDagTask reloadDagTask, DagProcessingEngine dagProcessingEngine) {
+    return new ReloadDagProc(reloadDagTask, dagProcessingEngine);
+  }


can we get rid of this one after previous discussion about no need to reload preemptively? we only access dagStateStore at time of execution

umustafi · 2024-01-03T23:52:21Z

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

+@Slf4j
+@AllArgsConstructor
+// change to iterable
+public class DagTaskStream implements Iterator<DagTask>{


the java doc mentions implementing DagManagement but after recent convo and seeing NewDagManager is the one to implement the code looks fine but update ja

umustafi · 2024-01-04T00:15:28Z

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

+ */
+@Slf4j
+@Alpha
+// todo - maybe reload need different treatment from other operations, because these are already present in the dag store


do we still need this? what is loading here? every time we start up we only need to initialize the scheduler then the scheduler should take care of triggering launch tasks. or if there are unfinished tasks in the dagActionStore those will be separately dealt with by the startup sequence.

arjun4084346 force-pushed the DagRefactor branch 7 times, most recently from 1ebc701 to 55b2ae2 Compare November 13, 2023 19:04

arjun4084346 force-pushed the DagRefactor branch from 97dd7b0 to 2b88493 Compare November 15, 2023 20:55

arjun4084346 force-pushed the DagRefactor branch 2 times, most recently from 8eac85d to 3845841 Compare December 4, 2023 22:43

umustafi reviewed Dec 14, 2023

View reviewed changes

arjun4084346 force-pushed the DagRefactor branch from 1b826a4 to 280ba6d Compare December 14, 2023 22:45

umustafi reviewed Dec 14, 2023

View reviewed changes

arjun4084346 force-pushed the DagRefactor branch from 280ba6d to eafad07 Compare December 15, 2023 00:17

umustafi reviewed Dec 15, 2023

View reviewed changes

arjun4084346 force-pushed the DagRefactor branch 2 times, most recently from 7416b31 to 48794e9 Compare December 19, 2023 01:29

phet reviewed Dec 19, 2023

View reviewed changes

arjun4084346 force-pushed the DagRefactor branch 5 times, most recently from 88affd9 to d45f8d5 Compare December 22, 2023 02:32

initial commit for refactoring dag manager

a9d3ca0

remove changes not related to dag manager refactoring commit before preemption use visiter pattern store dag action instead of dag task in dag task stream use dag management store in NewDagManager merge conflicts some cleaning fix tests

arjun4084346 force-pushed the DagRefactor branch from d45f8d5 to a9d3ca0 Compare December 22, 2023 23:06

umustafi reviewed Jan 1, 2024

View reviewed changes

umustafi reviewed Jan 4, 2024

View reviewed changes

arjun4084346 mentioned this pull request Jan 6, 2024

[GOBBLIN-1910] part one of the changes that dag refactoring will require #3853

Merged

4 tasks

arjun4084346 closed this Mar 13, 2024

		DagActionStore.DagAction dagAction = DagManagerUtils.createDagAction(flowGroup, flowName, String.valueOf(flowExecutionId), DagActionStore.FlowActionType.KILL);
		this.dagProcessingEngine.get().addDagAction(dagAction);

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



		/**
		* NewDagManager manages dags in memory and various mappings.

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

		import org.apache.gobblin.service.monitoring.event.JobStatusEvent;


		public interface DagManagement {

[draft] Dag refactor #3825

[draft] Dag refactor #3825

Conversation

arjun4084346 commented Nov 12, 2023

JIRA

Description

Tests

Commits

codecov-commenter commented Nov 13, 2023 • edited Loading

Codecov Report

umustafi 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

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

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

umustafi Dec 15, 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov-commenter commented Nov 13, 2023 •

edited

Loading

umustafi Dec 15, 2023 •

edited

Loading