You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@gobblin.apache.org by "arjun4084346 (via GitHub)" <gi...@apache.org> on 2024/07/12 02:12:58 UTC

[PR] implement DagNodeStateStore [gobblin]

arjun4084346 opened a new pull request, #3999:
URL: https://github.com/apache/gobblin/pull/3999

   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](https://issues.apache.org/jira/browse/GOBBLIN/) 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](http://chris.beams.io/posts/git-commit/)":
       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"
   
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1679996665


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementStateStore.java:
##########
@@ -66,25 +65,11 @@ public interface DagManagementStateStore {
    */
   void checkpointDag(Dag<JobExecutionPlan> dag) throws IOException;
 
-  /**
-   @return whether `dagId` is currently known due to {@link DagManagementStateStore#checkpointDag} but not yet
-   {@link DagManagementStateStore#deleteDag}
-   */
-  boolean containsDag(DagManager.DagId dagId) throws IOException;
-
   /**
    @return the {@link Dag}, if present
    */
   Optional<Dag<JobExecutionPlan>> getDag(DagManager.DagId dagId) throws IOException;
 
-  /**
-   * Delete the {@link Dag} from the backing store, typically called upon completion of execution.
-   * @param dag The dag completed/cancelled execution on {@link org.apache.gobblin.runtime.api.SpecExecutor}.
-   */
-  default void deleteDag(Dag<JobExecutionPlan> dag) throws IOException {
-    deleteDag(DagManagerUtils.generateDagId(dag));
-  }
-

Review Comment:
   why `deleteFailedDag(Dag<JEP>)` is preferable? if the caller has dag, it can create DagId, but if the caller only has dagId, it cannot create a Dag.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#issuecomment-2224310889

   ## [Codecov](https://app.codecov.io/gh/apache/gobblin/pull/3999?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
   All modified and coverable lines are covered by tests :white_check_mark:
   > Project coverage is 49.99%. Comparing base [(`b44a91e`)](https://app.codecov.io/gh/apache/gobblin/commit/b44a91ea69dded11fc2229237026a575ec94603a?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`918ac34`)](https://app.codecov.io/gh/apache/gobblin/commit/918ac34205216ee39dbc3212ba58bba8b1359e0d?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
   > Report is 2 commits behind head on master.
   
   <details><summary>Additional details and impacted files</summary>
   
   
   ```diff
   @@              Coverage Diff              @@
   ##             master    #3999       +/-   ##
   =============================================
   + Coverage     38.72%   49.99%   +11.26%     
   - Complexity     1596     5872     +4276     
   =============================================
     Files           388     1074      +686     
     Lines         15957    41230    +25273     
     Branches       1578     4625     +3047     
   =============================================
   + Hits           6180    20611    +14431     
   - Misses         9283    18831     +9548     
   - Partials        494     1788     +1294     
   ```
   
   </details>
   
   [:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/gobblin/pull/3999?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).   
   :loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1675030784


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementStateStore.java:
##########
@@ -66,25 +65,11 @@ public interface DagManagementStateStore {
    */
   void checkpointDag(Dag<JobExecutionPlan> dag) throws IOException;
 
-  /**
-   @return whether `dagId` is currently known due to {@link DagManagementStateStore#checkpointDag} but not yet
-   {@link DagManagementStateStore#deleteDag}
-   */

Review Comment:
   removing the unused APIs



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2115] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1683385416


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/LaunchDagProc.java:
##########
@@ -84,8 +84,6 @@ protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag
       // todo - add metrics
     } else {
       DagProcUtils.submitNextNodes(dagManagementStateStore, dag.get(), getDagId());
-      // Checkpoint the dag state, it should have an updated value of dag nodes

Review Comment:
   because DagProcUtils.submitNextNodes updates the table already



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1680058204


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2.java:
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagNodeStateStore} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreV2 implements DagNodeStateStore {
+
+  public static final String CONFIG_PREFIX = GOBBLIN_SERVICE_PREFIX + "mysqlDagStateStore";
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id))";

Review Comment:
   adding index. you suggested that dag_node_id alone is sufficient for PK



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1680122092


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -90,7 +90,7 @@ protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag
     dagManagementStateStore.checkpointDag(failedDag.get());
 
     // if it fails here, it will check point the failed dag in the (running) dag store again, which is idempotent
-    dagManagementStateStore.deleteFailedDag(failedDag.get());
+    dagManagementStateStore.deleteFailedDag(getDagId());

Review Comment:
   I think we should expect this kind of symmetry. It is not like a rest resource, where we do `get resource` and `delete resource`.
   It is a table with <dagId, dag>. Consider it like a `Map<K, V>` with auto incrementing key, we do not call delete<V> just because we do add(V).
   There are many other examples like, `Stack`'s API. 
   Basically, I believe, a `key` (dagId/dagNodeId) is a best way to choose a resource/row/dag in this case, not the dag.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2115] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 merged PR #3999:
URL: https://github.com/apache/gobblin/pull/3999


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1675035833


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStore.java:
##########
@@ -121,10 +115,10 @@ public synchronized void setTopologySpecMap(Map<URI, TopologySpec> topologySpecM
     start();
   }
 
-  private DagStateStore createDagStateStore(Config config, Map<URI, TopologySpec> topologySpecMap) {
+  private DagNodeStateStore createDagStateStore(Config config, Map<URI, TopologySpec> topologySpecMap) {
     try {
       Class<?> dagStateStoreClass = Class.forName(ConfigUtils.getString(config, DAG_STATESTORE_CLASS_KEY, MysqlDagStateStore.class.getName()));

Review Comment:
   Using dag proc engine is a pre requisite of using MysqlDagStateStoreV2, so keeping the default class `MysqlDagStateStore`. though it is also set to MysqlDagStateStore through configs in LI projects



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2115] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1683382853


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreWithDagNodes.java:
##########
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagStateStoreWithDagNodes} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreWithDagNodes implements DagStateStoreWithDagNodes {
+
+  public static final String CONFIG_PREFIX = MysqlDagStateStore.CONFIG_PREFIX;
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON NOT NULL, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id), "
+      + "UNIQUE INDEX dag_node_index (dag_node_id), "
+      + "INDEX dag_index (parent_dag_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_NODES_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";
+  protected static final String GET_DAG_NODE_STATEMENT = "SELECT dag_node FROM %s WHERE dag_node_id = ?";
+  protected static final String DELETE_DAG_STATEMENT = "DELETE FROM %s WHERE parent_dag_id = ?";
+  private final ContextAwareCounter totalDagCount;
+
+  public MysqlDagStateStoreWithDagNodes(Config config, Map<URI, TopologySpec> topologySpecMap) throws IOException {
+    if (config.hasPath(CONFIG_PREFIX)) {
+      config = config.getConfig(CONFIG_PREFIX).withFallback(config);
+    }
+
+    String DEFAULT_TABLE_NAME = "dag_node_state_store";
+    this.tableName = ConfigUtils.getString(config, ConfigurationKeys.STATE_STORE_DB_TABLE_KEY, DEFAULT_TABLE_NAME);
+    // create table if it does not exist
+    DataSource dataSource = MysqlDataSourceFactory.get(config, SharedResourcesBrokerFactory.getImplicitBroker());
+
+    try (Connection connection = dataSource.getConnection();
+        PreparedStatement createStatement = connection.prepareStatement(String.format(CREATE_TABLE_STATEMENT, tableName))) {
+      createStatement.executeUpdate();
+      connection.commit();
+    } catch (SQLException e) {
+      throw new IOException("Failure creation table " + tableName, e);
+    }
+    this.dbStatementExecutor = new DBStatementExecutor(dataSource, log);
+
+    JsonSerializer<List<JobExecutionPlan>> serializer = new JobExecutionPlanListSerializer();
+    JsonDeserializer<List<JobExecutionPlan>> deserializer = new JobExecutionPlanListDeserializer(topologySpecMap);
+    Type typeToken = new TypeToken<List<JobExecutionPlan>>() {
+    }.getType();
+    this.serDe = new GsonSerDe<>(typeToken, serializer, deserializer);
+    this.jobExecPlanDagFactory = new JobExecutionPlanDagFactory();
+    MetricContext metricContext =
+        Instrumented.getMetricContext(new State(ConfigUtils.configToProperties(config)), this.getClass());
+    this.totalDagCount = metricContext.contextAwareCounter(ServiceMetricNames.DAG_COUNT_MYSQL_DAG_STATE_COUNT);
+    log.info("Instantiated {}", getClass().getSimpleName());
+  }
+
+  @Override
+  public void writeCheckpoint(Dag<JobExecutionPlan> dag)
+      throws IOException {
+    DagManager.DagId dagId = DagManagerUtils.generateDagId(dag);
+    boolean newDag = false;
+    for (Dag.DagNode<JobExecutionPlan> dagNode : dag.getNodes()) {
+      if (updateDagNode(dagId, dagNode) == 1) {
+        newDag = true;
+      }
+    }
+    if (newDag) {
+      this.totalDagCount.inc();
+    }
+  }
+
+  @Override
+  public void cleanUp(Dag<JobExecutionPlan> dag) throws IOException {
+    cleanUp(generateDagId(dag));
+  }
+
+  @Override
+  public boolean cleanUp(DagManager.DagId dagId) throws IOException {
+    dbStatementExecutor.withPreparedStatement(String.format(DELETE_DAG_STATEMENT, tableName), deleteStatement -> {
+      try {
+        deleteStatement.setString(1, dagId.toString());
+        return deleteStatement.executeUpdate() != 0;
+      } catch (SQLException e) {
+        throw new IOException(String.format("Failure deleting dag for %s", dagId), e);
+      }}, true);
+    this.totalDagCount.dec();
+    return true;
+  }
+
+  @Override
+  public void cleanUp(String dagId) throws IOException {
+    throw new NotSupportedException(getClass().getSimpleName() + " does not need this legacy API that originated with "
+        + "the DagManager that is replaced by DagProcessingEngine");
+  }
+
+  @Override
+  public List<Dag<JobExecutionPlan>> getDags() throws IOException {
+    throw new NotSupportedException(getClass().getSimpleName() + " does not need this legacy API that originated with "
+        + "the DagManager that is replaced by DagProcessingEngine");  }
+
+  @Override
+  public Dag<JobExecutionPlan> getDag(DagManager.DagId dagId) throws IOException {
+    return convertDagNodesIntoDag(getDagNodes(dagId));
+  }
+
+  @Override
+  public Dag<JobExecutionPlan> getDag(String dagId) throws IOException {
+    throw new NotSupportedException(getClass().getSimpleName() + " does not need this API");
+  }
+
+  @Override
+  public Set<String> getDagIds() throws IOException {
+    throw new NotSupportedException(getClass().getSimpleName() + " does not need this API");
+  }
+
+  /**
+   * Get the {@link Dag} out of a {@link State} pocket.
+   */
+  private Dag<JobExecutionPlan> convertDagNodesIntoDag(Set<Dag.DagNode<JobExecutionPlan>> dagNodes) {
+    if (dagNodes.isEmpty()) {
+      return null;

Review Comment:
   as per kip's comments, it is too late to change api signature for `DagStateStore.getDag` method, which is the only caller for this...



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1681808638


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStore.java:
##########
@@ -116,128 +112,78 @@ public void removeFlowSpec(URI uri, Properties headers, boolean triggerListener)
     this.flowCatalog.remove(uri, headers, triggerListener);
   }
 
-  public synchronized void setTopologySpecMap(Map<URI, TopologySpec> topologySpecMap) throws IOException {
+  public synchronized void setTopologySpecMap(Map<URI, TopologySpec> topologySpecMap) {
     this.topologySpecMap = topologySpecMap;
     start();
   }
 
-  private DagStateStore createDagStateStore(Config config, Map<URI, TopologySpec> topologySpecMap) {
+  private DagStateStoreWithDagNodes createDagStateStore(Config config, Map<URI, TopologySpec> topologySpecMap) {
     try {
-      Class<?> dagStateStoreClass = Class.forName(ConfigUtils.getString(config, DAG_STATESTORE_CLASS_KEY, MysqlDagStateStore.class.getName()));
-      return (DagStateStore) GobblinConstructorUtils.invokeLongestConstructor(dagStateStoreClass, config, topologySpecMap);
+      Class<?> dagStateStoreClass = Class.forName(ConfigUtils.getString(config, DAG_STATESTORE_CLASS_KEY, MysqlDagStateStoreWithDagNodes.class.getName()));
+      return (DagStateStoreWithDagNodes) GobblinConstructorUtils.invokeLongestConstructor(dagStateStoreClass, config, topologySpecMap);
     } catch (ReflectiveOperationException e) {
       throw new RuntimeException(e);
     }
   }
 
   @Override
-  public void checkpointDag(Dag<JobExecutionPlan> dag) throws IOException {

Review Comment:
   this api will no longer be used for "checkpointing", will be used just for adding the dag for once in the beginning for it's lifecycle (by launch and resume dag procs)



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1681809379


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/DagProcUtils.java:
##########
@@ -70,7 +70,6 @@ public static void submitNextNodes(DagManagementStateStore dagManagementStateSto
     if (nextNodes.size() == 1) {
       Dag.DagNode<JobExecutionPlan> dagNode = nextNodes.iterator().next();
       DagProcUtils.submitJobToExecutor(dagManagementStateStore, dagNode, dagId);
-      log.info("Submitted job {} for dagId {}", DagManagerUtils.getJobName(dagNode), dagId);

Review Comment:
   duplicate log line



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2115] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1683376782


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagStateStoreWithDagNodes.java:
##########
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.util.Optional;
+import java.util.Set;
+
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+
+/**
+ * An interface for storing and retrieving currently running {@link Dag.DagNode<JobExecutionPlan>}s.
+ * Callers should use {@link DagStateStore#writeCheckpoint} to store dags. After that, to update individual
+ * {@link Dag.DagNode}s, {@link DagStateStoreWithDagNodes#updateDagNode} should be used.
+ * {@link DagStateStore#cleanUp(DagManager.DagId)} should be used to delete all the {@link Dag.DagNode}s for a {@link Dag}.
+ */
+public interface DagStateStoreWithDagNodes extends DagStateStore {
+  int updateDagNode(DagManager.DagId dagId, Dag.DagNode<JobExecutionPlan> dagNode) throws IOException;
+
+  Set<Dag.DagNode<JobExecutionPlan>> getDagNodes(DagManager.DagId dagId) throws IOException;
+
+  Optional<Dag.DagNode<JobExecutionPlan>> getDagNode(DagNodeId dagNodeId) throws IOException;

Review Comment:
   because the set can be empty



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1680066676


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2.java:
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagNodeStateStore} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreV2 implements DagNodeStateStore {
+
+  public static final String CONFIG_PREFIX = GOBBLIN_SERVICE_PREFIX + "mysqlDagStateStore";
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";
+  protected static final String GET_DAG_NODE_STATEMENT = "SELECT dag_node FROM %s WHERE dag_node_id = ?";
+  protected static final String DELETE_DAG_STATEMENT = "DELETE FROM %s WHERE parent_dag_id = ?";
+  protected static final String DELETE_DAG_NODE_STATEMENT = "DELETE FROM %s WHERE dag_node_id = ?";
+  private final ContextAwareCounter totalDagCount;
+
+  public MysqlDagStateStoreV2(Config config, Map<URI, TopologySpec> topologySpecMap) throws IOException {
+    if (config.hasPath(CONFIG_PREFIX)) {
+      config = config.getConfig(CONFIG_PREFIX).withFallback(config);
+    }
+
+    String DEFAULT_TABLE_NAME = "dag_node_state_store";
+    this.tableName = ConfigUtils.getString(config, ConfigurationKeys.STATE_STORE_DB_TABLE_KEY, DEFAULT_TABLE_NAME);
+    // create table if it does not exist
+    DataSource dataSource = MysqlDataSourceFactory.get(config, SharedResourcesBrokerFactory.getImplicitBroker());
+
+    try (Connection connection = dataSource.getConnection();
+        PreparedStatement createStatement = connection.prepareStatement(String.format(CREATE_TABLE_STATEMENT, tableName))) {
+      createStatement.executeUpdate();
+      connection.commit();
+    } catch (SQLException e) {
+      throw new IOException("Failure creation table " + tableName, e);
+    }
+    this.dbStatementExecutor = new DBStatementExecutor(dataSource, log);
+
+    JsonSerializer<List<JobExecutionPlan>> serializer = new JobExecutionPlanListSerializer();
+    JsonDeserializer<List<JobExecutionPlan>> deserializer = new JobExecutionPlanListDeserializer(topologySpecMap);
+    Type typeToken = new TypeToken<List<JobExecutionPlan>>() {
+    }.getType();
+    this.serDe = new GsonSerDe<>(typeToken, serializer, deserializer);
+    this.jobExecPlanDagFactory = new JobExecutionPlanDagFactory();
+    MetricContext metricContext =
+        Instrumented.getMetricContext(new State(ConfigUtils.configToProperties(config)), this.getClass());
+    this.totalDagCount = metricContext.contextAwareCounter(ServiceMetricNames.DAG_COUNT_MYSQL_DAG_STATE_COUNT);
+
+  }
+
+  @Override
+  public void writeCheckpoint(Dag<JobExecutionPlan> dag)
+      throws IOException {
+    DagManager.DagId dagId = DagManagerUtils.generateDagId(dag);
+    boolean newDag = false;
+    for (Dag.DagNode<JobExecutionPlan> dagNode : dag.getNodes()) {
+      if (addDagNodeState(dagNode, dagId) == 1) {
+        newDag = true;
+      }
+    }
+    if (newDag) {
+      this.totalDagCount.inc();
+    }
+  }
+
+  @Override
+  public void cleanUp(Dag<JobExecutionPlan> dag) throws IOException {
+    cleanUp(generateDagId(dag).toString());
+  }
+
+  @Override
+  public boolean cleanUp(DagManager.DagId dagId) throws IOException {
+    dbStatementExecutor.withPreparedStatement(String.format(DELETE_DAG_STATEMENT, tableName), deleteStatement -> {
+      try {
+        deleteStatement.setString(1, dagId.toString());

Review Comment:
   1 is the count of `?` in the prepared statement



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1676168069


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/core/GobblinServiceGuiceModule.java:
##########
@@ -178,7 +178,7 @@ public void configure(Binder binder) {
     OptionalBinder.newOptionalBinder(binder, DagActionStore.class);
     if (serviceConfig.isWarmStandbyEnabled()) {
       binder.bind(DagActionStore.class).to(MysqlDagActionStore.class);
-      binder.bind(DagManagementStateStore.class).to(MostlyMySqlDagManagementStateStore.class);
+      binder.bind(DagManagementStateStore.class).to(MySqlDagManagementStateStore.class);

Review Comment:
   do we only want this switched for dagProcessingEngine or in general?



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2.java:
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagNodeStateStore} using MySQL as a backup.

Review Comment:
   can you outline the differences between here and v1?



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2.java:
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagNodeStateStore} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreV2 implements DagNodeStateStore {
+
+  public static final String CONFIG_PREFIX = GOBBLIN_SERVICE_PREFIX + "mysqlDagStateStore";
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";
+  protected static final String GET_DAG_NODE_STATEMENT = "SELECT dag_node FROM %s WHERE dag_node_id = ?";
+  protected static final String DELETE_DAG_STATEMENT = "DELETE FROM %s WHERE parent_dag_id = ?";
+  protected static final String DELETE_DAG_NODE_STATEMENT = "DELETE FROM %s WHERE dag_node_id = ?";
+  private final ContextAwareCounter totalDagCount;
+
+  public MysqlDagStateStoreV2(Config config, Map<URI, TopologySpec> topologySpecMap) throws IOException {
+    if (config.hasPath(CONFIG_PREFIX)) {
+      config = config.getConfig(CONFIG_PREFIX).withFallback(config);
+    }
+
+    String DEFAULT_TABLE_NAME = "dag_node_state_store";
+    this.tableName = ConfigUtils.getString(config, ConfigurationKeys.STATE_STORE_DB_TABLE_KEY, DEFAULT_TABLE_NAME);
+    // create table if it does not exist
+    DataSource dataSource = MysqlDataSourceFactory.get(config, SharedResourcesBrokerFactory.getImplicitBroker());
+
+    try (Connection connection = dataSource.getConnection();
+        PreparedStatement createStatement = connection.prepareStatement(String.format(CREATE_TABLE_STATEMENT, tableName))) {
+      createStatement.executeUpdate();
+      connection.commit();
+    } catch (SQLException e) {
+      throw new IOException("Failure creation table " + tableName, e);
+    }
+    this.dbStatementExecutor = new DBStatementExecutor(dataSource, log);
+
+    JsonSerializer<List<JobExecutionPlan>> serializer = new JobExecutionPlanListSerializer();
+    JsonDeserializer<List<JobExecutionPlan>> deserializer = new JobExecutionPlanListDeserializer(topologySpecMap);
+    Type typeToken = new TypeToken<List<JobExecutionPlan>>() {
+    }.getType();
+    this.serDe = new GsonSerDe<>(typeToken, serializer, deserializer);
+    this.jobExecPlanDagFactory = new JobExecutionPlanDagFactory();
+    MetricContext metricContext =
+        Instrumented.getMetricContext(new State(ConfigUtils.configToProperties(config)), this.getClass());
+    this.totalDagCount = metricContext.contextAwareCounter(ServiceMetricNames.DAG_COUNT_MYSQL_DAG_STATE_COUNT);
+
+  }
+
+  @Override
+  public void writeCheckpoint(Dag<JobExecutionPlan> dag)
+      throws IOException {
+    DagManager.DagId dagId = DagManagerUtils.generateDagId(dag);
+    boolean newDag = false;
+    for (Dag.DagNode<JobExecutionPlan> dagNode : dag.getNodes()) {
+      if (addDagNodeState(dagNode, dagId) == 1) {
+        newDag = true;
+      }
+    }
+    if (newDag) {
+      this.totalDagCount.inc();
+    }
+  }
+
+  @Override
+  public void cleanUp(Dag<JobExecutionPlan> dag) throws IOException {
+    cleanUp(generateDagId(dag).toString());
+  }
+
+  @Override
+  public boolean cleanUp(DagManager.DagId dagId) throws IOException {
+    dbStatementExecutor.withPreparedStatement(String.format(DELETE_DAG_STATEMENT, tableName), deleteStatement -> {
+      try {
+        deleteStatement.setString(1, dagId.toString());

Review Comment:
   why index 1? you already completed the first `%s` with the tableName in the `String.format` about



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2.java:
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagNodeStateStore} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreV2 implements DagNodeStateStore {
+
+  public static final String CONFIG_PREFIX = GOBBLIN_SERVICE_PREFIX + "mysqlDagStateStore";
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";
+  protected static final String GET_DAG_NODE_STATEMENT = "SELECT dag_node FROM %s WHERE dag_node_id = ?";
+  protected static final String DELETE_DAG_STATEMENT = "DELETE FROM %s WHERE parent_dag_id = ?";
+  protected static final String DELETE_DAG_NODE_STATEMENT = "DELETE FROM %s WHERE dag_node_id = ?";
+  private final ContextAwareCounter totalDagCount;
+
+  public MysqlDagStateStoreV2(Config config, Map<URI, TopologySpec> topologySpecMap) throws IOException {
+    if (config.hasPath(CONFIG_PREFIX)) {
+      config = config.getConfig(CONFIG_PREFIX).withFallback(config);
+    }
+
+    String DEFAULT_TABLE_NAME = "dag_node_state_store";
+    this.tableName = ConfigUtils.getString(config, ConfigurationKeys.STATE_STORE_DB_TABLE_KEY, DEFAULT_TABLE_NAME);
+    // create table if it does not exist
+    DataSource dataSource = MysqlDataSourceFactory.get(config, SharedResourcesBrokerFactory.getImplicitBroker());
+
+    try (Connection connection = dataSource.getConnection();
+        PreparedStatement createStatement = connection.prepareStatement(String.format(CREATE_TABLE_STATEMENT, tableName))) {
+      createStatement.executeUpdate();
+      connection.commit();
+    } catch (SQLException e) {
+      throw new IOException("Failure creation table " + tableName, e);
+    }
+    this.dbStatementExecutor = new DBStatementExecutor(dataSource, log);
+
+    JsonSerializer<List<JobExecutionPlan>> serializer = new JobExecutionPlanListSerializer();
+    JsonDeserializer<List<JobExecutionPlan>> deserializer = new JobExecutionPlanListDeserializer(topologySpecMap);
+    Type typeToken = new TypeToken<List<JobExecutionPlan>>() {
+    }.getType();
+    this.serDe = new GsonSerDe<>(typeToken, serializer, deserializer);
+    this.jobExecPlanDagFactory = new JobExecutionPlanDagFactory();
+    MetricContext metricContext =
+        Instrumented.getMetricContext(new State(ConfigUtils.configToProperties(config)), this.getClass());
+    this.totalDagCount = metricContext.contextAwareCounter(ServiceMetricNames.DAG_COUNT_MYSQL_DAG_STATE_COUNT);
+
+  }
+
+  @Override
+  public void writeCheckpoint(Dag<JobExecutionPlan> dag)
+      throws IOException {
+    DagManager.DagId dagId = DagManagerUtils.generateDagId(dag);
+    boolean newDag = false;
+    for (Dag.DagNode<JobExecutionPlan> dagNode : dag.getNodes()) {
+      if (addDagNodeState(dagNode, dagId) == 1) {
+        newDag = true;
+      }
+    }
+    if (newDag) {
+      this.totalDagCount.inc();
+    }
+  }
+
+  @Override
+  public void cleanUp(Dag<JobExecutionPlan> dag) throws IOException {
+    cleanUp(generateDagId(dag).toString());
+  }
+
+  @Override
+  public boolean cleanUp(DagManager.DagId dagId) throws IOException {
+    dbStatementExecutor.withPreparedStatement(String.format(DELETE_DAG_STATEMENT, tableName), deleteStatement -> {
+      try {
+        deleteStatement.setString(1, dagId.toString());
+        return deleteStatement.executeUpdate() != 0;
+      } catch (SQLException e) {
+        throw new IOException(String.format("Failure deleting dag for %s", dagId), e);
+      }}, true);
+    this.totalDagCount.dec();
+    return true;
+  }
+
+  @Override
+  public void cleanUp(String dagId) throws IOException {
+    throw new NotSupportedException(getClass().getSimpleName() + " does not need this API");
+  }
+
+  @Override
+  public List<Dag<JobExecutionPlan>> getDags() throws IOException {
+    throw new NotSupportedException(getClass().getSimpleName() + " does not need this API");
+  }
+
+  @Override
+  public Dag<JobExecutionPlan> getDag(DagManager.DagId dagId) throws IOException {
+    return convertDagNodesIntoDag(getDagNodes(dagId));
+  }
+
+  @Override
+  public Dag<JobExecutionPlan> getDag(String dagId) throws IOException {
+    throw new NotSupportedException(getClass().getSimpleName() + " does not need this API");
+  }
+
+  @Override
+  public Set<String> getDagIds() throws IOException {
+    throw new NotSupportedException(getClass().getSimpleName() + " does not need this API");
+  }
+
+  /**
+   * Get the {@link Dag} out of a {@link State} pocket.
+   */
+  private Dag<JobExecutionPlan> convertDagNodesIntoDag(Set<Dag.DagNode<JobExecutionPlan>> dagNodes) {
+    if (dagNodes.isEmpty()) {
+      return null;
+    }
+    return jobExecPlanDagFactory.createDag(dagNodes.stream().map(Dag.DagNode::getValue).collect(Collectors.toList()));
+  }
+
+  @Override
+  public int addDagNodeState(Dag.DagNode<JobExecutionPlan> dagNode, DagManager.DagId parentDagId) throws IOException {
+    String dagNodeId = dagNode.getValue().getId().toString();
+    return dbStatementExecutor.withPreparedStatement(String.format(INSERT_STATEMENT, tableName), insertStatement -> {
+      try {
+        insertStatement.setString(1, dagNodeId);
+        insertStatement.setString(2, parentDagId.toString());
+        insertStatement.setString(3, this.serDe.serialize(Collections.singletonList(dagNode.getValue())));
+        return insertStatement.executeUpdate();

Review Comment:
   double check about the indices being right here (same comment as above)



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStore.java:
##########


Review Comment:
   I agree we may want to be able to switch between the two implementations 



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2.java:
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagNodeStateStore} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreV2 implements DagNodeStateStore {
+
+  public static final String CONFIG_PREFIX = GOBBLIN_SERVICE_PREFIX + "mysqlDagStateStore";
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id))";

Review Comment:
   in the doc I thought u wanted an index on `parent_dag_id` and that it is part of the PK (although I don't think its necessary but making it part of PK may help make the index. Also can `dag_node` be NULL? Why? Perhaps also add a comment to explain what that column corresponds to



##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2Test.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import com.typesafe.config.Config;
+
+import org.apache.gobblin.config.ConfigBuilder;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.metastore.testing.ITestMetastoreDatabase;
+import org.apache.gobblin.metastore.testing.TestMetastoreDatabaseFactory;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+
+/**
+ * Mainly testing functionalities related to DagStateStore but not Mysql-related components.
+ */
+public class MysqlDagStateStoreV2Test {
+
+  private DagStateStore dagStateStore;
+
+  private static final String TEST_USER = "testUser";
+  private static ITestMetastoreDatabase testDb;
+
+  @BeforeClass
+  public void setUp() throws Exception {
+    testDb = TestMetastoreDatabaseFactory.get();
+    ConfigBuilder configBuilder = ConfigBuilder.create()
+        .addPrimitive(ConfigurationKeys.STATE_STORE_DB_USER_KEY, TEST_USER)
+        .addPrimitive(ConfigurationKeys.STATE_STORE_DB_URL_KEY, testDb.getJdbcUrl())
+        .addPrimitive(ConfigurationKeys.STATE_STORE_DB_PASSWORD_KEY, MySqlDagManagementStateStoreTest.TEST_PASSWORD);
+
+    // Constructing TopologySpecMap.
+    Map<URI, TopologySpec> topologySpecMap = new HashMap<>();
+    String specExecInstance = "mySpecExecutor";
+    TopologySpec topologySpec = DagTestUtils.buildNaiveTopologySpec(specExecInstance);
+    URI specExecURI = new URI(specExecInstance);
+    topologySpecMap.put(specExecURI, topologySpec);
+    this.dagStateStore = new MysqlDagStateStoreV2(configBuilder.build(), topologySpecMap);
+  }
+
+  @AfterClass(alwaysRun = true)
+  public void tearDown() throws Exception {
+    if (testDb != null) {
+      // `.close()` to avoid (in the aggregate, across multiple suites) - java.sql.SQLNonTransientConnectionException: Too many connections
+      testDb.close();
+    }
+  }
+
+  @Test
+  public void testWriteGetAndDeleteDag() throws Exception{
+    Dag<JobExecutionPlan> dag_0 = DagTestUtils.buildDag("random_0", 123L);
+    Dag<JobExecutionPlan> dag_1 = DagTestUtils.buildDag("random_1", 456L);
+    DagManager.DagId dagId0 = DagManagerUtils.generateDagId(dag_0);
+    DagManager.DagId dagId1 = DagManagerUtils.generateDagId(dag_1);
+    this.dagStateStore.writeCheckpoint(dag_0);
+    this.dagStateStore.writeCheckpoint(dag_1);
+
+    // Verify get one dag
+    Dag<JobExecutionPlan> dag0 = this.dagStateStore.getDag(dagId0);
+    Dag<JobExecutionPlan> dag1 = this.dagStateStore.getDag(dagId1);
+    Assert.assertTrue(MySqlDagManagementStateStoreTest.compareLists(dag0.getNodes(), dag_0.getNodes()));
+    Assert.assertTrue(MySqlDagManagementStateStoreTest.compareLists(dag1.getNodes(), dag_1.getNodes()));
+
+    // Verify dag contents
+    Dag<JobExecutionPlan> dagDeserialized = dag0;
+    Assert.assertEquals(dagDeserialized.getNodes().size(), 2);
+    Assert.assertEquals(dagDeserialized.getStartNodes().size(), 1);
+    Assert.assertEquals(dagDeserialized.getEndNodes().size(), 1);
+    Dag.DagNode<JobExecutionPlan> child = dagDeserialized.getEndNodes().get(0);
+    Dag.DagNode<JobExecutionPlan> parent = dagDeserialized.getStartNodes().get(0);
+    Assert.assertEquals(dagDeserialized.getParentChildMap().size(), 1);
+    Assert.assertTrue(dagDeserialized.getParentChildMap().get(parent).contains(child));
+
+    for (int i = 0; i < 2; i++) {
+      JobExecutionPlan plan = dagDeserialized.getNodes().get(i).getValue();
+      Config jobConfig = plan.getJobSpec().getConfig();
+      Assert.assertEquals(jobConfig.getString(ConfigurationKeys.FLOW_GROUP_KEY), "group" + "random_0");
+      Assert.assertEquals(jobConfig.getString(ConfigurationKeys.FLOW_NAME_KEY), "flow" + "random_0");
+      Assert.assertEquals(jobConfig.getLong(ConfigurationKeys.FLOW_EXECUTION_ID_KEY), 123L);
+      Assert.assertEquals(plan.getExecutionStatus(), ExecutionStatus.RUNNING);
+      Assert.assertTrue(Boolean.parseBoolean(plan.getJobFuture().get().get().toString()));
+      Assert.assertTrue(Boolean.parseBoolean(plan.getJobFuture().get().get().toString()));

Review Comment:
   could maybe replace this common test functionality with a method



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2.java:
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagNodeStateStore} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreV2 implements DagNodeStateStore {
+
+  public static final String CONFIG_PREFIX = GOBBLIN_SERVICE_PREFIX + "mysqlDagStateStore";
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";

Review Comment:
   won't this retrieve multiple? I agree let's rename it to show it will get plural
   `GET_DAG_NODES_BY_PARENT_ID`



##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2Test.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import com.typesafe.config.Config;
+
+import org.apache.gobblin.config.ConfigBuilder;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.metastore.testing.ITestMetastoreDatabase;
+import org.apache.gobblin.metastore.testing.TestMetastoreDatabaseFactory;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+
+/**
+ * Mainly testing functionalities related to DagStateStore but not Mysql-related components.
+ */
+public class MysqlDagStateStoreV2Test {
+
+  private DagStateStore dagStateStore;
+
+  private static final String TEST_USER = "testUser";
+  private static ITestMetastoreDatabase testDb;
+
+  @BeforeClass
+  public void setUp() throws Exception {
+    testDb = TestMetastoreDatabaseFactory.get();
+    ConfigBuilder configBuilder = ConfigBuilder.create()
+        .addPrimitive(ConfigurationKeys.STATE_STORE_DB_USER_KEY, TEST_USER)
+        .addPrimitive(ConfigurationKeys.STATE_STORE_DB_URL_KEY, testDb.getJdbcUrl())
+        .addPrimitive(ConfigurationKeys.STATE_STORE_DB_PASSWORD_KEY, MySqlDagManagementStateStoreTest.TEST_PASSWORD);
+
+    // Constructing TopologySpecMap.
+    Map<URI, TopologySpec> topologySpecMap = new HashMap<>();
+    String specExecInstance = "mySpecExecutor";
+    TopologySpec topologySpec = DagTestUtils.buildNaiveTopologySpec(specExecInstance);
+    URI specExecURI = new URI(specExecInstance);
+    topologySpecMap.put(specExecURI, topologySpec);
+    this.dagStateStore = new MysqlDagStateStoreV2(configBuilder.build(), topologySpecMap);
+  }
+
+  @AfterClass(alwaysRun = true)
+  public void tearDown() throws Exception {
+    if (testDb != null) {
+      // `.close()` to avoid (in the aggregate, across multiple suites) - java.sql.SQLNonTransientConnectionException: Too many connections
+      testDb.close();
+    }
+  }
+
+  @Test
+  public void testWriteGetAndDeleteDag() throws Exception{

Review Comment:
   addGetAndDelete?



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2.java:
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagNodeStateStore} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreV2 implements DagNodeStateStore {
+
+  public static final String CONFIG_PREFIX = GOBBLIN_SERVICE_PREFIX + "mysqlDagStateStore";
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";
+  protected static final String GET_DAG_NODE_STATEMENT = "SELECT dag_node FROM %s WHERE dag_node_id = ?";
+  protected static final String DELETE_DAG_STATEMENT = "DELETE FROM %s WHERE parent_dag_id = ?";
+  protected static final String DELETE_DAG_NODE_STATEMENT = "DELETE FROM %s WHERE dag_node_id = ?";
+  private final ContextAwareCounter totalDagCount;
+
+  public MysqlDagStateStoreV2(Config config, Map<URI, TopologySpec> topologySpecMap) throws IOException {
+    if (config.hasPath(CONFIG_PREFIX)) {
+      config = config.getConfig(CONFIG_PREFIX).withFallback(config);
+    }
+
+    String DEFAULT_TABLE_NAME = "dag_node_state_store";
+    this.tableName = ConfigUtils.getString(config, ConfigurationKeys.STATE_STORE_DB_TABLE_KEY, DEFAULT_TABLE_NAME);
+    // create table if it does not exist
+    DataSource dataSource = MysqlDataSourceFactory.get(config, SharedResourcesBrokerFactory.getImplicitBroker());
+
+    try (Connection connection = dataSource.getConnection();
+        PreparedStatement createStatement = connection.prepareStatement(String.format(CREATE_TABLE_STATEMENT, tableName))) {
+      createStatement.executeUpdate();
+      connection.commit();
+    } catch (SQLException e) {
+      throw new IOException("Failure creation table " + tableName, e);
+    }
+    this.dbStatementExecutor = new DBStatementExecutor(dataSource, log);
+
+    JsonSerializer<List<JobExecutionPlan>> serializer = new JobExecutionPlanListSerializer();
+    JsonDeserializer<List<JobExecutionPlan>> deserializer = new JobExecutionPlanListDeserializer(topologySpecMap);
+    Type typeToken = new TypeToken<List<JobExecutionPlan>>() {
+    }.getType();
+    this.serDe = new GsonSerDe<>(typeToken, serializer, deserializer);
+    this.jobExecPlanDagFactory = new JobExecutionPlanDagFactory();
+    MetricContext metricContext =
+        Instrumented.getMetricContext(new State(ConfigUtils.configToProperties(config)), this.getClass());
+    this.totalDagCount = metricContext.contextAwareCounter(ServiceMetricNames.DAG_COUNT_MYSQL_DAG_STATE_COUNT);
+

Review Comment:
   add log at the end of the constructor after it's been initialized



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2.java:
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagNodeStateStore} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreV2 implements DagNodeStateStore {
+
+  public static final String CONFIG_PREFIX = GOBBLIN_SERVICE_PREFIX + "mysqlDagStateStore";
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";
+  protected static final String GET_DAG_NODE_STATEMENT = "SELECT dag_node FROM %s WHERE dag_node_id = ?";
+  protected static final String DELETE_DAG_STATEMENT = "DELETE FROM %s WHERE parent_dag_id = ?";
+  protected static final String DELETE_DAG_NODE_STATEMENT = "DELETE FROM %s WHERE dag_node_id = ?";
+  private final ContextAwareCounter totalDagCount;
+
+  public MysqlDagStateStoreV2(Config config, Map<URI, TopologySpec> topologySpecMap) throws IOException {
+    if (config.hasPath(CONFIG_PREFIX)) {
+      config = config.getConfig(CONFIG_PREFIX).withFallback(config);
+    }
+
+    String DEFAULT_TABLE_NAME = "dag_node_state_store";
+    this.tableName = ConfigUtils.getString(config, ConfigurationKeys.STATE_STORE_DB_TABLE_KEY, DEFAULT_TABLE_NAME);
+    // create table if it does not exist
+    DataSource dataSource = MysqlDataSourceFactory.get(config, SharedResourcesBrokerFactory.getImplicitBroker());
+
+    try (Connection connection = dataSource.getConnection();
+        PreparedStatement createStatement = connection.prepareStatement(String.format(CREATE_TABLE_STATEMENT, tableName))) {
+      createStatement.executeUpdate();
+      connection.commit();
+    } catch (SQLException e) {
+      throw new IOException("Failure creation table " + tableName, e);
+    }
+    this.dbStatementExecutor = new DBStatementExecutor(dataSource, log);
+
+    JsonSerializer<List<JobExecutionPlan>> serializer = new JobExecutionPlanListSerializer();
+    JsonDeserializer<List<JobExecutionPlan>> deserializer = new JobExecutionPlanListDeserializer(topologySpecMap);
+    Type typeToken = new TypeToken<List<JobExecutionPlan>>() {
+    }.getType();
+    this.serDe = new GsonSerDe<>(typeToken, serializer, deserializer);
+    this.jobExecPlanDagFactory = new JobExecutionPlanDagFactory();
+    MetricContext metricContext =
+        Instrumented.getMetricContext(new State(ConfigUtils.configToProperties(config)), this.getClass());
+    this.totalDagCount = metricContext.contextAwareCounter(ServiceMetricNames.DAG_COUNT_MYSQL_DAG_STATE_COUNT);
+
+  }
+
+  @Override
+  public void writeCheckpoint(Dag<JobExecutionPlan> dag)
+      throws IOException {
+    DagManager.DagId dagId = DagManagerUtils.generateDagId(dag);
+    boolean newDag = false;
+    for (Dag.DagNode<JobExecutionPlan> dagNode : dag.getNodes()) {
+      if (addDagNodeState(dagNode, dagId) == 1) {
+        newDag = true;
+      }
+    }
+    if (newDag) {
+      this.totalDagCount.inc();
+    }
+  }
+
+  @Override
+  public void cleanUp(Dag<JobExecutionPlan> dag) throws IOException {
+    cleanUp(generateDagId(dag).toString());
+  }
+
+  @Override
+  public boolean cleanUp(DagManager.DagId dagId) throws IOException {
+    dbStatementExecutor.withPreparedStatement(String.format(DELETE_DAG_STATEMENT, tableName), deleteStatement -> {
+      try {
+        deleteStatement.setString(1, dagId.toString());

Review Comment:
   want a unit test for this if one not there



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1680317198


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -125,7 +125,7 @@ public Orchestrator(Config config, TopologyCatalog topologyCatalog, DagManager d
     //At this point, the TopologySpecMap is initialized by the SpecCompiler. Pass the TopologySpecMap to the DagManager.
     this.dagManager.setTopologySpecMap(getSpecCompiler().getTopologySpecMap());
     if (dagManagementStateStore.isPresent()) {
-      ((MostlyMySqlDagManagementStateStore) dagManagementStateStore.get()).setTopologySpecMap(getSpecCompiler().getTopologySpecMap());
+      ((MySqlDagManagementStateStore) dagManagementStateStore.get()).setTopologySpecMap(getSpecCompiler().getTopologySpecMap());

Review Comment:
   because i believe setting topology should happen in constructor. thinking of creating a ticket for this for future team



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1680122092


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -90,7 +90,7 @@ protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag
     dagManagementStateStore.checkpointDag(failedDag.get());
 
     // if it fails here, it will check point the failed dag in the (running) dag store again, which is idempotent
-    dagManagementStateStore.deleteFailedDag(failedDag.get());
+    dagManagementStateStore.deleteFailedDag(getDagId());

Review Comment:
   I think we should expect this kind of symmetry. It is not like a rest resource, where we do `get resource` and `delete resource`.
   It is a table with <dagId, dag>. Consider it like a auto incrementing key `Map<K, V>`, we do not call delete<V> just because we do add(V).
   There are many other examples like, `Stack`'s API. 
   Basically, I believe, a `key` (dagId/dagNodeId) is a best way to choose a resource/row/dag in this case, not the dag.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2115] implement DagNodeStateStore [gobblin]

Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1683256692


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagStateStoreWithDagNodes.java:
##########
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.util.Optional;
+import java.util.Set;
+
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+
+/**
+ * An interface for storing and retrieving currently running {@link Dag.DagNode<JobExecutionPlan>}s.
+ * Callers should use {@link DagStateStore#writeCheckpoint} to store dags. After that, to update individual
+ * {@link Dag.DagNode}s, {@link DagStateStoreWithDagNodes#updateDagNode} should be used.
+ * {@link DagStateStore#cleanUp(DagManager.DagId)} should be used to delete all the {@link Dag.DagNode}s for a {@link Dag}.
+ */
+public interface DagStateStoreWithDagNodes extends DagStateStore {
+  int updateDagNode(DagManager.DagId dagId, Dag.DagNode<JobExecutionPlan> dagNode) throws IOException;
+
+  Set<Dag.DagNode<JobExecutionPlan>> getDagNodes(DagManager.DagId dagId) throws IOException;
+
+  Optional<Dag.DagNode<JobExecutionPlan>> getDagNode(DagNodeId dagNodeId) throws IOException;

Review Comment:
   why is this one optional but the set above is not? Is it bc dagNodeId may not exist or be cleaned up?



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementStateStore.java:
##########
@@ -42,6 +41,8 @@
  * 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.
  */
+// todo - this should merge with DagStateStoreWithDagNodes interface and `addDagNodeState` and `addDagNode` should merge and rename to

Review Comment:
   does `DagStateStoreWithDagNodes` not extend `DagManagementStateStore`? If not then why?



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/DagProcUtils.java:
##########
@@ -169,8 +169,8 @@ public static void cancelDagNode(Dag.DagNode<JobExecutionPlan> dagNodeToCancel,
             dagNodeToCancel.getValue().getJobSpec().getUri());
       }
       DagManagerUtils.getSpecProducer(dagNodeToCancel).cancelJob(dagNodeToCancel.getValue().getJobSpec().getUri(), props).get();
-      // todo - why was it not being cleaned up in DagManager?
-      dagManagementStateStore.deleteDagNodeState(dagId, dagNodeToCancel);
+      // add back the dag node with updated states in the store
+      dagManagementStateStore.addDagNodeState(dagNodeToCancel, dagId);

Review Comment:
   does this go in failedDagStateStore later once processing the cancellation event?



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagStateStoreWithDagNodes.java:
##########
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.util.Optional;
+import java.util.Set;
+
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+
+/**
+ * An interface for storing and retrieving currently running {@link Dag.DagNode<JobExecutionPlan>}s.
+ * Callers should use {@link DagStateStore#writeCheckpoint} to store dags. After that, to update individual
+ * {@link Dag.DagNode}s, {@link DagStateStoreWithDagNodes#updateDagNode} should be used.
+ * {@link DagStateStore#cleanUp(DagManager.DagId)} should be used to delete all the {@link Dag.DagNode}s for a {@link Dag}.
+ */
+public interface DagStateStoreWithDagNodes extends DagStateStore {
+  int updateDagNode(DagManager.DagId dagId, Dag.DagNode<JobExecutionPlan> dagNode) throws IOException;

Review Comment:
   java doc to explain what the return type here is and also add another new line



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreWithDagNodes.java:
##########
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagStateStoreWithDagNodes} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreWithDagNodes implements DagStateStoreWithDagNodes {
+
+  public static final String CONFIG_PREFIX = MysqlDagStateStore.CONFIG_PREFIX;
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON NOT NULL, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id), "
+      + "UNIQUE INDEX dag_node_index (dag_node_id), "
+      + "INDEX dag_index (parent_dag_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_NODES_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";
+  protected static final String GET_DAG_NODE_STATEMENT = "SELECT dag_node FROM %s WHERE dag_node_id = ?";
+  protected static final String DELETE_DAG_STATEMENT = "DELETE FROM %s WHERE parent_dag_id = ?";
+  private final ContextAwareCounter totalDagCount;
+
+  public MysqlDagStateStoreWithDagNodes(Config config, Map<URI, TopologySpec> topologySpecMap) throws IOException {
+    if (config.hasPath(CONFIG_PREFIX)) {
+      config = config.getConfig(CONFIG_PREFIX).withFallback(config);
+    }
+
+    String DEFAULT_TABLE_NAME = "dag_node_state_store";
+    this.tableName = ConfigUtils.getString(config, ConfigurationKeys.STATE_STORE_DB_TABLE_KEY, DEFAULT_TABLE_NAME);
+    // create table if it does not exist
+    DataSource dataSource = MysqlDataSourceFactory.get(config, SharedResourcesBrokerFactory.getImplicitBroker());
+
+    try (Connection connection = dataSource.getConnection();
+        PreparedStatement createStatement = connection.prepareStatement(String.format(CREATE_TABLE_STATEMENT, tableName))) {
+      createStatement.executeUpdate();
+      connection.commit();
+    } catch (SQLException e) {
+      throw new IOException("Failure creation table " + tableName, e);
+    }
+    this.dbStatementExecutor = new DBStatementExecutor(dataSource, log);
+
+    JsonSerializer<List<JobExecutionPlan>> serializer = new JobExecutionPlanListSerializer();
+    JsonDeserializer<List<JobExecutionPlan>> deserializer = new JobExecutionPlanListDeserializer(topologySpecMap);
+    Type typeToken = new TypeToken<List<JobExecutionPlan>>() {
+    }.getType();
+    this.serDe = new GsonSerDe<>(typeToken, serializer, deserializer);
+    this.jobExecPlanDagFactory = new JobExecutionPlanDagFactory();
+    MetricContext metricContext =
+        Instrumented.getMetricContext(new State(ConfigUtils.configToProperties(config)), this.getClass());
+    this.totalDagCount = metricContext.contextAwareCounter(ServiceMetricNames.DAG_COUNT_MYSQL_DAG_STATE_COUNT);
+    log.info("Instantiated {}", getClass().getSimpleName());
+  }
+
+  @Override
+  public void writeCheckpoint(Dag<JobExecutionPlan> dag)

Review Comment:
   I thought we won't use this anymore in favor of updateDag?



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreWithDagNodes.java:
##########
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagStateStoreWithDagNodes} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreWithDagNodes implements DagStateStoreWithDagNodes {
+
+  public static final String CONFIG_PREFIX = MysqlDagStateStore.CONFIG_PREFIX;
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON NOT NULL, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id), "
+      + "UNIQUE INDEX dag_node_index (dag_node_id), "
+      + "INDEX dag_index (parent_dag_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_NODES_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";
+  protected static final String GET_DAG_NODE_STATEMENT = "SELECT dag_node FROM %s WHERE dag_node_id = ?";
+  protected static final String DELETE_DAG_STATEMENT = "DELETE FROM %s WHERE parent_dag_id = ?";
+  private final ContextAwareCounter totalDagCount;
+
+  public MysqlDagStateStoreWithDagNodes(Config config, Map<URI, TopologySpec> topologySpecMap) throws IOException {
+    if (config.hasPath(CONFIG_PREFIX)) {
+      config = config.getConfig(CONFIG_PREFIX).withFallback(config);
+    }
+
+    String DEFAULT_TABLE_NAME = "dag_node_state_store";

Review Comment:
   put this above as a class field



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStore.java:
##########
@@ -62,13 +60,11 @@
  */
 @Slf4j
 @Singleton
-public class MostlyMySqlDagManagementStateStore implements DagManagementStateStore {

Review Comment:
   where are the sql statements for this class? to create/update the store



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/LaunchDagProc.java:
##########
@@ -84,8 +84,6 @@ protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag
       // todo - add metrics
     } else {
       DagProcUtils.submitNextNodes(dagManagementStateStore, dag.get(), getDagId());
-      // Checkpoint the dag state, it should have an updated value of dag nodes

Review Comment:
   why do we not update the store here? is it done in conclude or only needs to be done once in initialize?



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ReevaluateDagProc.java:
##########
@@ -167,9 +160,5 @@ private void onJobFinish(DagManagementStateStore dagManagementStateStore, Dag.Da
       default:
         log.warn("It should not reach here. Job status {} is unexpected.", executionStatus);
     }
-
-    // Checkpoint the dag state, it should have an updated value of dag fields
-    dagManagementStateStore.checkpointDag(dag);

Review Comment:
   is this removed bc the dag was not updated in this method?



##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStoreTest.java:
##########
@@ -90,41 +103,40 @@ public void testAddDag() throws Exception {
     DagManager.DagId dagId2 = DagManagerUtils.generateDagId(dag2);
     DagNodeId dagNodeId = DagManagerUtils.calcJobId(dagNode.getValue().getJobSpec().getConfig());
 
-    this.dagManagementStateStore.checkpointDag(dag);
-    this.dagManagementStateStore.checkpointDag(dag2);
+    this.dagManagementStateStore.addDag(dag);
+    this.dagManagementStateStore.addDag(dag2);
     this.dagManagementStateStore.addDagNodeState(dagNode, dagId);
     this.dagManagementStateStore.addDagNodeState(dagNode2, dagId);
     this.dagManagementStateStore.addDagNodeState(dagNode3, dagId2);
 
-    Assert.assertTrue(this.dagManagementStateStore.containsDag(dagId));
-    Assert.assertEquals(dag.toString(), this.dagManagementStateStore.getDag(dagId).get().toString());
+    Assert.assertTrue(compareLists(dag.getNodes(), this.dagManagementStateStore.getDag(dagId).get().getNodes()));
     Assert.assertEquals(dagNode, this.dagManagementStateStore.getDagNodeWithJobStatus(dagNodeId).getLeft().get());
 
     Set<Dag.DagNode<JobExecutionPlan>> dagNodes = this.dagManagementStateStore.getDagNodes(dagId);
     Assert.assertEquals(2, dagNodes.size());
     Assert.assertTrue(dagNodes.contains(dagNode));
     Assert.assertTrue(dagNodes.contains(dagNode2));
 
-    this.dagManagementStateStore.deleteDagNodeState(dagId, dagNode);
-    Assert.assertFalse(this.dagManagementStateStore.getDagNodes(dagId).contains(dagNode));
-    Assert.assertTrue(this.dagManagementStateStore.getDagNodes(dagId).contains(dagNode2));
-    Assert.assertTrue(this.dagManagementStateStore.getDagNodes(dagId2).contains(dagNode3));
-
     // test to verify that adding a new dag node with the same dag node id (defined by the jobSpec) replaces the existing one
-    Assert.assertEquals(this.dagManagementStateStore.getDagNodes(dagId).size(), 1);

Review Comment:
   do we not want to assert the size before adding the duplicate?



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreWithDagNodes.java:
##########
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagStateStoreWithDagNodes} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreWithDagNodes implements DagStateStoreWithDagNodes {
+
+  public static final String CONFIG_PREFIX = MysqlDagStateStore.CONFIG_PREFIX;
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON NOT NULL, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id), "
+      + "UNIQUE INDEX dag_node_index (dag_node_id), "
+      + "INDEX dag_index (parent_dag_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_NODES_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";
+  protected static final String GET_DAG_NODE_STATEMENT = "SELECT dag_node FROM %s WHERE dag_node_id = ?";
+  protected static final String DELETE_DAG_STATEMENT = "DELETE FROM %s WHERE parent_dag_id = ?";
+  private final ContextAwareCounter totalDagCount;
+
+  public MysqlDagStateStoreWithDagNodes(Config config, Map<URI, TopologySpec> topologySpecMap) throws IOException {
+    if (config.hasPath(CONFIG_PREFIX)) {
+      config = config.getConfig(CONFIG_PREFIX).withFallback(config);
+    }
+
+    String DEFAULT_TABLE_NAME = "dag_node_state_store";
+    this.tableName = ConfigUtils.getString(config, ConfigurationKeys.STATE_STORE_DB_TABLE_KEY, DEFAULT_TABLE_NAME);
+    // create table if it does not exist
+    DataSource dataSource = MysqlDataSourceFactory.get(config, SharedResourcesBrokerFactory.getImplicitBroker());
+
+    try (Connection connection = dataSource.getConnection();
+        PreparedStatement createStatement = connection.prepareStatement(String.format(CREATE_TABLE_STATEMENT, tableName))) {
+      createStatement.executeUpdate();
+      connection.commit();
+    } catch (SQLException e) {
+      throw new IOException("Failure creation table " + tableName, e);
+    }
+    this.dbStatementExecutor = new DBStatementExecutor(dataSource, log);
+
+    JsonSerializer<List<JobExecutionPlan>> serializer = new JobExecutionPlanListSerializer();
+    JsonDeserializer<List<JobExecutionPlan>> deserializer = new JobExecutionPlanListDeserializer(topologySpecMap);
+    Type typeToken = new TypeToken<List<JobExecutionPlan>>() {
+    }.getType();
+    this.serDe = new GsonSerDe<>(typeToken, serializer, deserializer);
+    this.jobExecPlanDagFactory = new JobExecutionPlanDagFactory();
+    MetricContext metricContext =
+        Instrumented.getMetricContext(new State(ConfigUtils.configToProperties(config)), this.getClass());
+    this.totalDagCount = metricContext.contextAwareCounter(ServiceMetricNames.DAG_COUNT_MYSQL_DAG_STATE_COUNT);
+    log.info("Instantiated {}", getClass().getSimpleName());
+  }
+
+  @Override
+  public void writeCheckpoint(Dag<JobExecutionPlan> dag)
+      throws IOException {
+    DagManager.DagId dagId = DagManagerUtils.generateDagId(dag);
+    boolean newDag = false;
+    for (Dag.DagNode<JobExecutionPlan> dagNode : dag.getNodes()) {
+      if (updateDagNode(dagId, dagNode) == 1) {
+        newDag = true;
+      }
+    }
+    if (newDag) {
+      this.totalDagCount.inc();
+    }
+  }
+
+  @Override
+  public void cleanUp(Dag<JobExecutionPlan> dag) throws IOException {
+    cleanUp(generateDagId(dag));
+  }
+
+  @Override
+  public boolean cleanUp(DagManager.DagId dagId) throws IOException {
+    dbStatementExecutor.withPreparedStatement(String.format(DELETE_DAG_STATEMENT, tableName), deleteStatement -> {
+      try {
+        deleteStatement.setString(1, dagId.toString());
+        return deleteStatement.executeUpdate() != 0;
+      } catch (SQLException e) {
+        throw new IOException(String.format("Failure deleting dag for %s", dagId), e);
+      }}, true);
+    this.totalDagCount.dec();
+    return true;
+  }
+
+  @Override
+  public void cleanUp(String dagId) throws IOException {
+    throw new NotSupportedException(getClass().getSimpleName() + " does not need this legacy API that originated with "
+        + "the DagManager that is replaced by DagProcessingEngine");
+  }
+
+  @Override
+  public List<Dag<JobExecutionPlan>> getDags() throws IOException {
+    throw new NotSupportedException(getClass().getSimpleName() + " does not need this legacy API that originated with "
+        + "the DagManager that is replaced by DagProcessingEngine");  }
+
+  @Override
+  public Dag<JobExecutionPlan> getDag(DagManager.DagId dagId) throws IOException {
+    return convertDagNodesIntoDag(getDagNodes(dagId));
+  }
+
+  @Override
+  public Dag<JobExecutionPlan> getDag(String dagId) throws IOException {
+    throw new NotSupportedException(getClass().getSimpleName() + " does not need this API");
+  }
+
+  @Override
+  public Set<String> getDagIds() throws IOException {
+    throw new NotSupportedException(getClass().getSimpleName() + " does not need this API");
+  }
+
+  /**
+   * Get the {@link Dag} out of a {@link State} pocket.
+   */
+  private Dag<JobExecutionPlan> convertDagNodesIntoDag(Set<Dag.DagNode<JobExecutionPlan>> dagNodes) {
+    if (dagNodes.isEmpty()) {
+      return null;

Review Comment:
   its better to use an optional here so we know to check for null



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementStateStore.java:
##########
@@ -66,25 +65,11 @@ public interface DagManagementStateStore {
    */
   void checkpointDag(Dag<JobExecutionPlan> dag) throws IOException;
 
-  /**
-   @return whether `dagId` is currently known due to {@link DagManagementStateStore#checkpointDag} but not yet
-   {@link DagManagementStateStore#deleteDag}
-   */
-  boolean containsDag(DagManager.DagId dagId) throws IOException;
-
   /**
    @return the {@link Dag}, if present
    */
   Optional<Dag<JobExecutionPlan>> getDag(DagManager.DagId dagId) throws IOException;
 
-  /**
-   * Delete the {@link Dag} from the backing store, typically called upon completion of execution.
-   * @param dag The dag completed/cancelled execution on {@link org.apache.gobblin.runtime.api.SpecExecutor}.
-   */
-  default void deleteDag(Dag<JobExecutionPlan> dag) throws IOException {
-    deleteDag(DagManagerUtils.generateDagId(dag));
-  }
-

Review Comment:
   By your logic, then isn't this method necessary in a situation only dagId is available not the jobExecutionPlan?



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStore.java:
##########


Review Comment:
   I see we want to always fix forward then



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagStateStoreWithDagNodes.java:
##########
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.util.Optional;
+import java.util.Set;
+
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+
+/**
+ * An interface for storing and retrieving currently running {@link Dag.DagNode<JobExecutionPlan>}s.
+ * Callers should use {@link DagStateStore#writeCheckpoint} to store dags. After that, to update individual
+ * {@link Dag.DagNode}s, {@link DagStateStoreWithDagNodes#updateDagNode} should be used.
+ * {@link DagStateStore#cleanUp(DagManager.DagId)} should be used to delete all the {@link Dag.DagNode}s for a {@link Dag}.
+ */
+public interface DagStateStoreWithDagNodes extends DagStateStore {
+  int updateDagNode(DagManager.DagId dagId, Dag.DagNode<JobExecutionPlan> dagNode) throws IOException;
+
+  Set<Dag.DagNode<JobExecutionPlan>> getDagNodes(DagManager.DagId dagId) throws IOException;
+
+  Optional<Dag.DagNode<JobExecutionPlan>> getDagNode(DagNodeId dagNodeId) throws IOException;

Review Comment:
   java doc would help



##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreWithDagNodesTest.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import com.typesafe.config.Config;
+
+import org.apache.gobblin.config.ConfigBuilder;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.metastore.testing.ITestMetastoreDatabase;
+import org.apache.gobblin.metastore.testing.TestMetastoreDatabaseFactory;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+
+/**
+ * Mainly testing functionalities related to DagStateStore but not Mysql-related components.
+ */
+public class MysqlDagStateStoreWithDagNodesTest {
+
+  private DagStateStore dagStateStore;
+
+  private static final String TEST_USER = "testUser";
+  private static ITestMetastoreDatabase testDb;
+
+  @BeforeClass
+  public void setUp() throws Exception {
+    testDb = TestMetastoreDatabaseFactory.get();
+    ConfigBuilder configBuilder = ConfigBuilder.create()
+        .addPrimitive(ConfigurationKeys.STATE_STORE_DB_USER_KEY, TEST_USER)
+        .addPrimitive(ConfigurationKeys.STATE_STORE_DB_URL_KEY, testDb.getJdbcUrl())
+        .addPrimitive(ConfigurationKeys.STATE_STORE_DB_PASSWORD_KEY, MySqlDagManagementStateStoreTest.TEST_PASSWORD);
+
+    // Constructing TopologySpecMap.
+    Map<URI, TopologySpec> topologySpecMap = new HashMap<>();
+    String specExecInstance = "mySpecExecutor";
+    TopologySpec topologySpec = DagTestUtils.buildNaiveTopologySpec(specExecInstance);
+    URI specExecURI = new URI(specExecInstance);
+    topologySpecMap.put(specExecURI, topologySpec);
+    this.dagStateStore = new MysqlDagStateStoreWithDagNodes(configBuilder.build(), topologySpecMap);
+  }
+
+  @AfterClass(alwaysRun = true)
+  public void tearDown() throws Exception {
+    if (testDb != null) {
+      // `.close()` to avoid (in the aggregate, across multiple suites) - java.sql.SQLNonTransientConnectionException: Too many connections
+      testDb.close();
+    }
+  }
+
+  @Test
+  public void testAddGetAndDeleteDag() throws Exception{
+    Dag<JobExecutionPlan> dag0Orig = DagTestUtils.buildDag("random_0", 123L);

Review Comment:
   make the name and id var so easy to assert later



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2115] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1683378900


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreWithDagNodes.java:
##########
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagStateStoreWithDagNodes} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreWithDagNodes implements DagStateStoreWithDagNodes {
+
+  public static final String CONFIG_PREFIX = MysqlDagStateStore.CONFIG_PREFIX;
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON NOT NULL, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id), "
+      + "UNIQUE INDEX dag_node_index (dag_node_id), "
+      + "INDEX dag_index (parent_dag_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_NODES_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";
+  protected static final String GET_DAG_NODE_STATEMENT = "SELECT dag_node FROM %s WHERE dag_node_id = ?";
+  protected static final String DELETE_DAG_STATEMENT = "DELETE FROM %s WHERE parent_dag_id = ?";
+  private final ContextAwareCounter totalDagCount;
+
+  public MysqlDagStateStoreWithDagNodes(Config config, Map<URI, TopologySpec> topologySpecMap) throws IOException {
+    if (config.hasPath(CONFIG_PREFIX)) {
+      config = config.getConfig(CONFIG_PREFIX).withFallback(config);
+    }
+
+    String DEFAULT_TABLE_NAME = "dag_node_state_store";

Review Comment:
   I did, but then intelliJ complains that it can be local because it is not used anywhere else.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2115] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1683480011


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreWithDagNodes.java:
##########
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagStateStoreWithDagNodes} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreWithDagNodes implements DagStateStoreWithDagNodes {
+
+  public static final String CONFIG_PREFIX = MysqlDagStateStore.CONFIG_PREFIX;
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON NOT NULL, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id), "
+      + "UNIQUE INDEX dag_node_index (dag_node_id), "
+      + "INDEX dag_index (parent_dag_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_NODES_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";
+  protected static final String GET_DAG_NODE_STATEMENT = "SELECT dag_node FROM %s WHERE dag_node_id = ?";
+  protected static final String DELETE_DAG_STATEMENT = "DELETE FROM %s WHERE parent_dag_id = ?";
+  private final ContextAwareCounter totalDagCount;
+
+  public MysqlDagStateStoreWithDagNodes(Config config, Map<URI, TopologySpec> topologySpecMap) throws IOException {
+    if (config.hasPath(CONFIG_PREFIX)) {
+      config = config.getConfig(CONFIG_PREFIX).withFallback(config);
+    }
+
+    String DEFAULT_TABLE_NAME = "dag_node_state_store";
+    this.tableName = ConfigUtils.getString(config, ConfigurationKeys.STATE_STORE_DB_TABLE_KEY, DEFAULT_TABLE_NAME);
+    // create table if it does not exist
+    DataSource dataSource = MysqlDataSourceFactory.get(config, SharedResourcesBrokerFactory.getImplicitBroker());
+
+    try (Connection connection = dataSource.getConnection();
+        PreparedStatement createStatement = connection.prepareStatement(String.format(CREATE_TABLE_STATEMENT, tableName))) {
+      createStatement.executeUpdate();
+      connection.commit();
+    } catch (SQLException e) {
+      throw new IOException("Failure creation table " + tableName, e);
+    }
+    this.dbStatementExecutor = new DBStatementExecutor(dataSource, log);
+
+    JsonSerializer<List<JobExecutionPlan>> serializer = new JobExecutionPlanListSerializer();
+    JsonDeserializer<List<JobExecutionPlan>> deserializer = new JobExecutionPlanListDeserializer(topologySpecMap);
+    Type typeToken = new TypeToken<List<JobExecutionPlan>>() {
+    }.getType();
+    this.serDe = new GsonSerDe<>(typeToken, serializer, deserializer);
+    this.jobExecPlanDagFactory = new JobExecutionPlanDagFactory();
+    MetricContext metricContext =
+        Instrumented.getMetricContext(new State(ConfigUtils.configToProperties(config)), this.getClass());
+    this.totalDagCount = metricContext.contextAwareCounter(ServiceMetricNames.DAG_COUNT_MYSQL_DAG_STATE_COUNT);
+    log.info("Instantiated {}", getClass().getSimpleName());
+  }
+
+  @Override
+  public void writeCheckpoint(Dag<JobExecutionPlan> dag)

Review Comment:
   as per kip's comment, it is too late to update name now for this or for parent interface DagStateStore. will come back to this after everything is stabilized and old classes/interfaces are removed. probably then we can rename it



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2115] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1683495917


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStore.java:
##########
@@ -116,128 +112,78 @@ public void removeFlowSpec(URI uri, Properties headers, boolean triggerListener)
     this.flowCatalog.remove(uri, headers, triggerListener);
   }
 
-  public synchronized void setTopologySpecMap(Map<URI, TopologySpec> topologySpecMap) throws IOException {
+  public synchronized void setTopologySpecMap(Map<URI, TopologySpec> topologySpecMap) {
     this.topologySpecMap = topologySpecMap;
     start();
   }
 
-  private DagStateStore createDagStateStore(Config config, Map<URI, TopologySpec> topologySpecMap) {
+  private DagStateStoreWithDagNodes createDagStateStore(Config config, Map<URI, TopologySpec> topologySpecMap) {
     try {
-      Class<?> dagStateStoreClass = Class.forName(ConfigUtils.getString(config, DAG_STATESTORE_CLASS_KEY, MysqlDagStateStore.class.getName()));
-      return (DagStateStore) GobblinConstructorUtils.invokeLongestConstructor(dagStateStoreClass, config, topologySpecMap);
+      Class<?> dagStateStoreClass = Class.forName(ConfigUtils.getString(config, DAG_STATESTORE_CLASS_KEY, MysqlDagStateStoreWithDagNodes.class.getName()));
+      return (DagStateStoreWithDagNodes) GobblinConstructorUtils.invokeLongestConstructor(dagStateStoreClass, config, topologySpecMap);
     } catch (ReflectiveOperationException e) {
       throw new RuntimeException(e);
     }
   }
 
   @Override
-  public void checkpointDag(Dag<JobExecutionPlan> dag) throws IOException {

Review Comment:
   actually with the new table, only dag nodes will be added in the table. (so far we add complete dag as one json)
   all the mutable operations happen only at dag node level.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1680067095


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2.java:
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagNodeStateStore} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreV2 implements DagNodeStateStore {
+
+  public static final String CONFIG_PREFIX = GOBBLIN_SERVICE_PREFIX + "mysqlDagStateStore";
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";
+  protected static final String GET_DAG_NODE_STATEMENT = "SELECT dag_node FROM %s WHERE dag_node_id = ?";
+  protected static final String DELETE_DAG_STATEMENT = "DELETE FROM %s WHERE parent_dag_id = ?";
+  protected static final String DELETE_DAG_NODE_STATEMENT = "DELETE FROM %s WHERE dag_node_id = ?";
+  private final ContextAwareCounter totalDagCount;
+
+  public MysqlDagStateStoreV2(Config config, Map<URI, TopologySpec> topologySpecMap) throws IOException {
+    if (config.hasPath(CONFIG_PREFIX)) {
+      config = config.getConfig(CONFIG_PREFIX).withFallback(config);
+    }
+
+    String DEFAULT_TABLE_NAME = "dag_node_state_store";
+    this.tableName = ConfigUtils.getString(config, ConfigurationKeys.STATE_STORE_DB_TABLE_KEY, DEFAULT_TABLE_NAME);
+    // create table if it does not exist
+    DataSource dataSource = MysqlDataSourceFactory.get(config, SharedResourcesBrokerFactory.getImplicitBroker());
+
+    try (Connection connection = dataSource.getConnection();
+        PreparedStatement createStatement = connection.prepareStatement(String.format(CREATE_TABLE_STATEMENT, tableName))) {
+      createStatement.executeUpdate();
+      connection.commit();
+    } catch (SQLException e) {
+      throw new IOException("Failure creation table " + tableName, e);
+    }
+    this.dbStatementExecutor = new DBStatementExecutor(dataSource, log);
+
+    JsonSerializer<List<JobExecutionPlan>> serializer = new JobExecutionPlanListSerializer();
+    JsonDeserializer<List<JobExecutionPlan>> deserializer = new JobExecutionPlanListDeserializer(topologySpecMap);
+    Type typeToken = new TypeToken<List<JobExecutionPlan>>() {
+    }.getType();
+    this.serDe = new GsonSerDe<>(typeToken, serializer, deserializer);
+    this.jobExecPlanDagFactory = new JobExecutionPlanDagFactory();
+    MetricContext metricContext =
+        Instrumented.getMetricContext(new State(ConfigUtils.configToProperties(config)), this.getClass());
+    this.totalDagCount = metricContext.contextAwareCounter(ServiceMetricNames.DAG_COUNT_MYSQL_DAG_STATE_COUNT);
+
+  }
+
+  @Override
+  public void writeCheckpoint(Dag<JobExecutionPlan> dag)
+      throws IOException {
+    DagManager.DagId dagId = DagManagerUtils.generateDagId(dag);
+    boolean newDag = false;
+    for (Dag.DagNode<JobExecutionPlan> dagNode : dag.getNodes()) {
+      if (addDagNodeState(dagNode, dagId) == 1) {
+        newDag = true;
+      }
+    }
+    if (newDag) {
+      this.totalDagCount.inc();
+    }
+  }
+
+  @Override
+  public void cleanUp(Dag<JobExecutionPlan> dag) throws IOException {
+    cleanUp(generateDagId(dag).toString());
+  }
+
+  @Override
+  public boolean cleanUp(DagManager.DagId dagId) throws IOException {
+    dbStatementExecutor.withPreparedStatement(String.format(DELETE_DAG_STATEMENT, tableName), deleteStatement -> {
+      try {
+        deleteStatement.setString(1, dagId.toString());

Review Comment:
   test is already present



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1680027766


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2.java:
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagNodeStateStore} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreV2 implements DagNodeStateStore {
+
+  public static final String CONFIG_PREFIX = GOBBLIN_SERVICE_PREFIX + "mysqlDagStateStore";
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "

Review Comment:
   utf8 takes 4 bytes for every character and making primary key column (dag_node_id) size beyond the allowed limit by MySQL
   changing to latin1 that takes 1 byte solved this problem



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1675415134


##########
gobblin-modules/gobblin-kafka-09/src/test/java/org/apache/gobblin/runtime/KafkaAvroJobStatusMonitorTest.java:
##########
@@ -197,7 +197,7 @@ public void testProcessMessageForSuccessfulFlow() throws IOException, Reflective
 
   @Test (dependsOnMethods = "testProcessMessageForSuccessfulFlow")
   public void testProcessMessageForFailedFlow() throws IOException, ReflectiveOperationException {
-    DagManagementStateStore dagManagementStateStore = mock(MostlyMySqlDagManagementStateStore.class);
+    DagManagementStateStore dagManagementStateStore = mock(MySqlDagManagementStateStore.class);

Review Comment:
   there isn't a difference between these two, is there?
   ```
   DagManagementStateStore dagManagementStateStore =
      mock(MySqlDagManagementStateStore.class);
   ```
   and
   ```
   DagManagementStateStore dagManagementStateStore =
       mock(DagManagementStateStore.class);
   ```
   
   I'd prefer the latter



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStore.java:
##########
@@ -136,103 +130,65 @@ public void checkpointDag(Dag<JobExecutionPlan> dag) throws IOException {
   }
 
   @Override
-  public void markDagFailed(Dag<JobExecutionPlan> dag) throws IOException {
-    this.dagStateStore.cleanUp(dag);
-    // todo - updated failedDagStateStore iff cleanup returned 1
+  public void markDagFailed(DagManager.DagId dagId) throws IOException {
+    Dag<JobExecutionPlan> dag = this.dagStateStore.getDag(dagId);
     this.failedDagStateStore.writeCheckpoint(dag);
-    log.info("Marked dag failed {}", DagManagerUtils.generateDagId(dag));
-  }
-
-  @Override
-  public void deleteDag(Dag<JobExecutionPlan> dag) throws IOException {
-    this.dagStateStore.cleanUp(dag);
-  }
-
-  @Override
-  public void deleteFailedDag(Dag<JobExecutionPlan> dag) throws IOException {
-    this.failedDagStateStore.cleanUp(dag);
+    this.dagStateStore.cleanUp(dagId);
+    // todo - updated failedDagStateStore iff cleanup returned 1
+    // or merge dagStateStore and failedDagStateStore and change the flag that marks a dag `failed`
+    log.info("Marked dag failed {}", dagId);
   }
 
   @Override
   public void deleteDag(DagManager.DagId dagId) throws IOException {
-    this.dagStateStore.cleanUp(dagId.toString());
+    this.dagStateStore.cleanUp(dagId);
     log.info("Deleted dag {}", dagId);
   }
 
   @Override
   public void deleteFailedDag(DagManager.DagId dagId) throws IOException {
-    this.failedDagStateStore.cleanUp(dagId.toString());
-  }
-
-  @Override
-  public List<Dag<JobExecutionPlan>> getDags() throws IOException {
-    return this.dagStateStore.getDags();
+    this.failedDagStateStore.cleanUp(dagId);

Review Comment:
   `deleteDag` logs the ID. should this one as well?



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagNodeStateStore.java:
##########
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.util.Optional;
+import java.util.Set;
+
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+
+/**
+ * An interface for storing and retrieving currently running {@link Dag.DagNode<JobExecutionPlan>}s.
+ */
+public interface DagNodeStateStore extends DagStateStore {

Review Comment:
   I wonder whether naming could make clearer this is still a `DagStateStore`, such as `NodeBasedDagStateStore` or `DagStateStoreWithDagNodes` or `DagNodeLevelDagStateStore`



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2.java:
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagNodeStateStore} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreV2 implements DagNodeStateStore {
+
+  public static final String CONFIG_PREFIX = GOBBLIN_SERVICE_PREFIX + "mysqlDagStateStore";
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";
+  protected static final String GET_DAG_NODE_STATEMENT = "SELECT dag_node FROM %s WHERE dag_node_id = ?";
+  protected static final String DELETE_DAG_STATEMENT = "DELETE FROM %s WHERE parent_dag_id = ?";
+  protected static final String DELETE_DAG_NODE_STATEMENT = "DELETE FROM %s WHERE dag_node_id = ?";
+  private final ContextAwareCounter totalDagCount;
+
+  public MysqlDagStateStoreV2(Config config, Map<URI, TopologySpec> topologySpecMap) throws IOException {
+    if (config.hasPath(CONFIG_PREFIX)) {
+      config = config.getConfig(CONFIG_PREFIX).withFallback(config);
+    }
+
+    String DEFAULT_TABLE_NAME = "dag_node_state_store";
+    this.tableName = ConfigUtils.getString(config, ConfigurationKeys.STATE_STORE_DB_TABLE_KEY, DEFAULT_TABLE_NAME);
+    // create table if it does not exist
+    DataSource dataSource = MysqlDataSourceFactory.get(config, SharedResourcesBrokerFactory.getImplicitBroker());
+
+    try (Connection connection = dataSource.getConnection();
+        PreparedStatement createStatement = connection.prepareStatement(String.format(CREATE_TABLE_STATEMENT, tableName))) {
+      createStatement.executeUpdate();
+      connection.commit();
+    } catch (SQLException e) {
+      throw new IOException("Failure creation table " + tableName, e);
+    }
+    this.dbStatementExecutor = new DBStatementExecutor(dataSource, log);
+
+    JsonSerializer<List<JobExecutionPlan>> serializer = new JobExecutionPlanListSerializer();
+    JsonDeserializer<List<JobExecutionPlan>> deserializer = new JobExecutionPlanListDeserializer(topologySpecMap);
+    Type typeToken = new TypeToken<List<JobExecutionPlan>>() {
+    }.getType();
+    this.serDe = new GsonSerDe<>(typeToken, serializer, deserializer);
+    this.jobExecPlanDagFactory = new JobExecutionPlanDagFactory();
+    MetricContext metricContext =
+        Instrumented.getMetricContext(new State(ConfigUtils.configToProperties(config)), this.getClass());
+    this.totalDagCount = metricContext.contextAwareCounter(ServiceMetricNames.DAG_COUNT_MYSQL_DAG_STATE_COUNT);
+
+  }
+
+  @Override
+  public void writeCheckpoint(Dag<JobExecutionPlan> dag)
+      throws IOException {
+    DagManager.DagId dagId = DagManagerUtils.generateDagId(dag);
+    boolean newDag = false;
+    for (Dag.DagNode<JobExecutionPlan> dagNode : dag.getNodes()) {
+      if (addDagNodeState(dagNode, dagId) == 1) {
+        newDag = true;
+      }
+    }
+    if (newDag) {
+      this.totalDagCount.inc();
+    }
+  }
+
+  @Override
+  public void cleanUp(Dag<JobExecutionPlan> dag) throws IOException {
+    cleanUp(generateDagId(dag).toString());
+  }
+
+  @Override
+  public boolean cleanUp(DagManager.DagId dagId) throws IOException {
+    dbStatementExecutor.withPreparedStatement(String.format(DELETE_DAG_STATEMENT, tableName), deleteStatement -> {
+      try {
+        deleteStatement.setString(1, dagId.toString());
+        return deleteStatement.executeUpdate() != 0;
+      } catch (SQLException e) {
+        throw new IOException(String.format("Failure deleting dag for %s", dagId), e);
+      }}, true);
+    this.totalDagCount.dec();
+    return true;
+  }
+
+  @Override
+  public void cleanUp(String dagId) throws IOException {
+    throw new NotSupportedException(getClass().getSimpleName() + " does not need this API");

Review Comment:
   think of future maintainers... explain why.
   
   e.g. "does not need this legacy API that originated with the DagManager" (?)



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagStateStore.java:
##########
@@ -47,28 +47,42 @@ public interface DagStateStore {
    */
   void cleanUp(Dag<JobExecutionPlan> dag) throws IOException;
 
+  default boolean cleanUp(DagManager.DagId dagId) throws IOException {
+    cleanUp(dagId.toString());
+    return true;
+  }
+
   /**
    * Delete the {@link Dag} from the backing store, typically upon completion of execution.
    * @param dagId The ID of the dag to clean up.
    */
+  // todo - change it to return boolean
+  @Deprecated

Review Comment:
   yes, plus it's unfortunately too late to change the API to return a boolean



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStore.java:
##########
@@ -62,13 +59,10 @@
  */
 @Slf4j
 @Singleton
-public class MostlyMySqlDagManagementStateStore implements DagManagementStateStore {
-  private final Map<DagNodeId, Dag.DagNode<JobExecutionPlan>> dagNodes = new ConcurrentHashMap<>();
-  // dagToJobs holds a map of dagId to running jobs of that dag
-  private final Map<DagManager.DagId, Set<Dag.DagNode<JobExecutionPlan>>> dagToJobs = new ConcurrentHashMap<>();
-  private DagStateStore dagStateStore;
-  private DagStateStore failedDagStateStore;
-  private JobStatusRetriever jobStatusRetriever;
+public class MySqlDagManagementStateStore implements DagManagementStateStore {

Review Comment:
   `Mysql` is traditionally capitalized as one word (rather than `MySql`).  we may have deviated w/ `MostlyMySqlDagMgmtSS`... but let's return to the typical convention `MysqlDagManagementSS` when we name this new class.
   
   that said, I don't actually see anything mysql-specific about this class (e.g. it doesn't directly hold a DB connection or use any SQL).  really there's only a config fallback:
   ```
   Class<?> dagStateStoreClass = Class.forName(ConfigUtils.getString(config, DAG_STATESTORE_CLASS_KEY, MysqlDagStateStore.class.getName()));
   ```
   
   maybe `DelegatingDagManagementStateStore` would be more accurate



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStore.java:
##########


Review Comment:
   given these are controlled by config, rather than editing the class in place to transform it, why not retain the `MySqlDagManagementStateStore`?  we could always remove that later.



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagNodeStateStore.java:
##########
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.util.Optional;
+import java.util.Set;
+
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+
+/**
+ * An interface for storing and retrieving currently running {@link Dag.DagNode<JobExecutionPlan>}s.
+ */
+public interface DagNodeStateStore extends DagStateStore {
+  int addDagNodeState(Dag.DagNode<JobExecutionPlan> dagNode, DagManager.DagId dagId) throws IOException;
+
+  boolean deleteDagNodeState(DagManager.DagId dagId, Dag.DagNode<JobExecutionPlan> dagNode) throws IOException;

Review Comment:
   suggest to use same param ordering for both, e.g. `(DM.DagId, Dag.DagNode<JEP>)`
   
   and how about simply `addDagNode` and `deleteDagNode` to parallel `getDagNode` and `getDagNodes`?



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -125,7 +125,7 @@ public Orchestrator(Config config, TopologyCatalog topologyCatalog, DagManager d
     //At this point, the TopologySpecMap is initialized by the SpecCompiler. Pass the TopologySpecMap to the DagManager.
     this.dagManager.setTopologySpecMap(getSpecCompiler().getTopologySpecMap());
     if (dagManagementStateStore.isPresent()) {
-      ((MostlyMySqlDagManagementStateStore) dagManagementStateStore.get()).setTopologySpecMap(getSpecCompiler().getTopologySpecMap());
+      ((MySqlDagManagementStateStore) dagManagementStateStore.get()).setTopologySpecMap(getSpecCompiler().getTopologySpecMap());

Review Comment:
   why do we need to cast this?  could/should `setTopoSpecMap` be part of the underlying interface, `DagManagementStateStore`?



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -90,7 +90,7 @@ protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag
     dagManagementStateStore.checkpointDag(failedDag.get());
 
     // if it fails here, it will check point the failed dag in the (running) dag store again, which is idempotent
-    dagManagementStateStore.deleteFailedDag(failedDag.get());
+    dagManagementStateStore.deleteFailedDag(getDagId());

Review Comment:
   I know it's essentially the same either way, but I do prefer the symmetry of:
   ```
   dmss.checkpointDag(failedDag.get());
   dmss.deleteFailedDag(failedDag.get());
   ```



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ReevaluateDagProc.java:
##########
@@ -114,7 +114,7 @@ protected void act(DagManagementStateStore dagManagementStateStore, Pair<Optiona
         // todo - verify if work from PR#3641 is required
         dagManagementStateStore.deleteDag(getDagId());
       } else {
-        dagManagementStateStore.markDagFailed(dag);
+        dagManagementStateStore.markDagFailed(getDagId());

Review Comment:
   if we still had the method, could this be:
   ```
   if (flowEvent.equals(TimingEvent.FlowTimings.FLOW_SUCCEEDED)) {
       // todo - verify if work from PR#3641 is required
       dagManagementStateStore.deleteDag(dag);
   } else {
       dagManagementStateStore.markDagFailed(dag);
   }
   ```
   ?



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2.java:
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagNodeStateStore} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreV2 implements DagNodeStateStore {
+
+  public static final String CONFIG_PREFIX = GOBBLIN_SERVICE_PREFIX + "mysqlDagStateStore";

Review Comment:
   is this reusing the config prefix of another class?  if so, let's be explicit:
   ```
   public static final String CONFIG_PREFIX = MysqlDagStateStore.CONFIG_PREFIX;
   ```
   



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementStateStore.java:
##########
@@ -66,25 +65,11 @@ public interface DagManagementStateStore {
    */
   void checkpointDag(Dag<JobExecutionPlan> dag) throws IOException;
 
-  /**
-   @return whether `dagId` is currently known due to {@link DagManagementStateStore#checkpointDag} but not yet
-   {@link DagManagementStateStore#deleteDag}
-   */
-  boolean containsDag(DagManager.DagId dagId) throws IOException;
-
   /**
    @return the {@link Dag}, if present
    */
   Optional<Dag<JobExecutionPlan>> getDag(DagManager.DagId dagId) throws IOException;
 
-  /**
-   * Delete the {@link Dag} from the backing store, typically called upon completion of execution.
-   * @param dag The dag completed/cancelled execution on {@link org.apache.gobblin.runtime.api.SpecExecutor}.
-   */
-  default void deleteDag(Dag<JobExecutionPlan> dag) throws IOException {
-    deleteDag(DagManagerUtils.generateDagId(dag));
-  }
-

Review Comment:
   I'm all for removing some of the unused methods, but this one and `deleteFailedDag(Dag<JEP>)` seem preferable abstraction to the equivalent form taking `DagManager.DagId`.  would it make sense for callers to use those instead (and encapsulate the `DMUtils.generateDagId(dag)` invocation)?



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2.java:
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagNodeStateStore} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreV2 implements DagNodeStateStore {
+
+  public static final String CONFIG_PREFIX = GOBBLIN_SERVICE_PREFIX + "mysqlDagStateStore";
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "

Review Comment:
   curious: why use `latin1`, rather than utf8 multi-byte?



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2.java:
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagNodeStateStore} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreV2 implements DagNodeStateStore {
+
+  public static final String CONFIG_PREFIX = GOBBLIN_SERVICE_PREFIX + "mysqlDagStateStore";
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";

Review Comment:
   `GET_DAG_NODES_STATEMENT`?  or `GET_DAG_NODES_BY_DAG_ID_STMT`?



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagStateStore.java:
##########
@@ -47,28 +47,42 @@ public interface DagStateStore {
    */
   void cleanUp(Dag<JobExecutionPlan> dag) throws IOException;
 
+  default boolean cleanUp(DagManager.DagId dagId) throws IOException {
+    cleanUp(dagId.toString());
+    return true;
+  }
+
   /**
    * Delete the {@link Dag} from the backing store, typically upon completion of execution.
    * @param dagId The ID of the dag to clean up.
    */
+  // todo - change it to return boolean
+  @Deprecated
   void cleanUp(String dagId) throws IOException;
 
   /**
    * Load all currently running {@link Dag}s from the underlying store. Typically, invoked when a new {@link DagManager}
    * takes over or on restart of service.
    * @return a {@link List} of currently running {@link Dag}s.
    */
+  @Deprecated

Review Comment:
   when deprecating, let's javadoc why - https://docs.oracle.com/javase/8/docs/technotes/guides/javadoc/deprecation/deprecation.html#javadoc_tag



##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2Test.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import com.typesafe.config.Config;
+
+import org.apache.gobblin.config.ConfigBuilder;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.metastore.testing.ITestMetastoreDatabase;
+import org.apache.gobblin.metastore.testing.TestMetastoreDatabaseFactory;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+
+/**
+ * Mainly testing functionalities related to DagStateStore but not Mysql-related components.
+ */
+public class MysqlDagStateStoreV2Test {
+
+  private DagStateStore dagStateStore;
+
+  private static final String TEST_USER = "testUser";
+  private static ITestMetastoreDatabase testDb;
+
+  @BeforeClass
+  public void setUp() throws Exception {
+    testDb = TestMetastoreDatabaseFactory.get();
+    ConfigBuilder configBuilder = ConfigBuilder.create()
+        .addPrimitive(ConfigurationKeys.STATE_STORE_DB_USER_KEY, TEST_USER)
+        .addPrimitive(ConfigurationKeys.STATE_STORE_DB_URL_KEY, testDb.getJdbcUrl())
+        .addPrimitive(ConfigurationKeys.STATE_STORE_DB_PASSWORD_KEY, MySqlDagManagementStateStoreTest.TEST_PASSWORD);
+
+    // Constructing TopologySpecMap.
+    Map<URI, TopologySpec> topologySpecMap = new HashMap<>();
+    String specExecInstance = "mySpecExecutor";
+    TopologySpec topologySpec = DagTestUtils.buildNaiveTopologySpec(specExecInstance);
+    URI specExecURI = new URI(specExecInstance);
+    topologySpecMap.put(specExecURI, topologySpec);
+    this.dagStateStore = new MysqlDagStateStoreV2(configBuilder.build(), topologySpecMap);
+  }
+
+  @AfterClass(alwaysRun = true)
+  public void tearDown() throws Exception {
+    if (testDb != null) {
+      // `.close()` to avoid (in the aggregate, across multiple suites) - java.sql.SQLNonTransientConnectionException: Too many connections
+      testDb.close();
+    }
+  }
+
+  @Test
+  public void testWriteGetAndDeleteDag() throws Exception{
+    Dag<JobExecutionPlan> dag_0 = DagTestUtils.buildDag("random_0", 123L);
+    Dag<JobExecutionPlan> dag_1 = DagTestUtils.buildDag("random_1", 456L);
+    DagManager.DagId dagId0 = DagManagerUtils.generateDagId(dag_0);
+    DagManager.DagId dagId1 = DagManagerUtils.generateDagId(dag_1);
+    this.dagStateStore.writeCheckpoint(dag_0);
+    this.dagStateStore.writeCheckpoint(dag_1);
+
+    // Verify get one dag
+    Dag<JobExecutionPlan> dag0 = this.dagStateStore.getDag(dagId0);
+    Dag<JobExecutionPlan> dag1 = this.dagStateStore.getDag(dagId1);
+    Assert.assertTrue(MySqlDagManagementStateStoreTest.compareLists(dag0.getNodes(), dag_0.getNodes()));
+    Assert.assertTrue(MySqlDagManagementStateStoreTest.compareLists(dag1.getNodes(), dag_1.getNodes()));
+
+    // Verify dag contents
+    Dag<JobExecutionPlan> dagDeserialized = dag0;
+    Assert.assertEquals(dagDeserialized.getNodes().size(), 2);
+    Assert.assertEquals(dagDeserialized.getStartNodes().size(), 1);
+    Assert.assertEquals(dagDeserialized.getEndNodes().size(), 1);
+    Dag.DagNode<JobExecutionPlan> child = dagDeserialized.getEndNodes().get(0);
+    Dag.DagNode<JobExecutionPlan> parent = dagDeserialized.getStartNodes().get(0);
+    Assert.assertEquals(dagDeserialized.getParentChildMap().size(), 1);
+    Assert.assertTrue(dagDeserialized.getParentChildMap().get(parent).contains(child));
+
+    for (int i = 0; i < 2; i++) {
+      JobExecutionPlan plan = dagDeserialized.getNodes().get(i).getValue();
+      Config jobConfig = plan.getJobSpec().getConfig();
+      Assert.assertEquals(jobConfig.getString(ConfigurationKeys.FLOW_GROUP_KEY), "group" + "random_0");
+      Assert.assertEquals(jobConfig.getString(ConfigurationKeys.FLOW_NAME_KEY), "flow" + "random_0");
+      Assert.assertEquals(jobConfig.getLong(ConfigurationKeys.FLOW_EXECUTION_ID_KEY), 123L);
+      Assert.assertEquals(plan.getExecutionStatus(), ExecutionStatus.RUNNING);
+      Assert.assertTrue(Boolean.parseBoolean(plan.getJobFuture().get().get().toString()));
+      Assert.assertTrue(Boolean.parseBoolean(plan.getJobFuture().get().get().toString()));
+    }
+
+    dagDeserialized = dag1;

Review Comment:
   nit: choose a fresh name, rather than re-assigning



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2.java:
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagNodeStateStore} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreV2 implements DagNodeStateStore {
+
+  public static final String CONFIG_PREFIX = GOBBLIN_SERVICE_PREFIX + "mysqlDagStateStore";
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";
+  protected static final String GET_DAG_NODE_STATEMENT = "SELECT dag_node FROM %s WHERE dag_node_id = ?";
+  protected static final String DELETE_DAG_STATEMENT = "DELETE FROM %s WHERE parent_dag_id = ?";
+  protected static final String DELETE_DAG_NODE_STATEMENT = "DELETE FROM %s WHERE dag_node_id = ?";
+  private final ContextAwareCounter totalDagCount;
+
+  public MysqlDagStateStoreV2(Config config, Map<URI, TopologySpec> topologySpecMap) throws IOException {
+    if (config.hasPath(CONFIG_PREFIX)) {
+      config = config.getConfig(CONFIG_PREFIX).withFallback(config);
+    }
+
+    String DEFAULT_TABLE_NAME = "dag_node_state_store";

Review Comment:
   should be `public static final`



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2.java:
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagNodeStateStore} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreV2 implements DagNodeStateStore {
+
+  public static final String CONFIG_PREFIX = GOBBLIN_SERVICE_PREFIX + "mysqlDagStateStore";
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";
+  protected static final String GET_DAG_NODE_STATEMENT = "SELECT dag_node FROM %s WHERE dag_node_id = ?";
+  protected static final String DELETE_DAG_STATEMENT = "DELETE FROM %s WHERE parent_dag_id = ?";
+  protected static final String DELETE_DAG_NODE_STATEMENT = "DELETE FROM %s WHERE dag_node_id = ?";
+  private final ContextAwareCounter totalDagCount;
+
+  public MysqlDagStateStoreV2(Config config, Map<URI, TopologySpec> topologySpecMap) throws IOException {
+    if (config.hasPath(CONFIG_PREFIX)) {
+      config = config.getConfig(CONFIG_PREFIX).withFallback(config);
+    }
+
+    String DEFAULT_TABLE_NAME = "dag_node_state_store";
+    this.tableName = ConfigUtils.getString(config, ConfigurationKeys.STATE_STORE_DB_TABLE_KEY, DEFAULT_TABLE_NAME);
+    // create table if it does not exist
+    DataSource dataSource = MysqlDataSourceFactory.get(config, SharedResourcesBrokerFactory.getImplicitBroker());
+
+    try (Connection connection = dataSource.getConnection();
+        PreparedStatement createStatement = connection.prepareStatement(String.format(CREATE_TABLE_STATEMENT, tableName))) {
+      createStatement.executeUpdate();
+      connection.commit();
+    } catch (SQLException e) {
+      throw new IOException("Failure creation table " + tableName, e);
+    }
+    this.dbStatementExecutor = new DBStatementExecutor(dataSource, log);
+
+    JsonSerializer<List<JobExecutionPlan>> serializer = new JobExecutionPlanListSerializer();
+    JsonDeserializer<List<JobExecutionPlan>> deserializer = new JobExecutionPlanListDeserializer(topologySpecMap);
+    Type typeToken = new TypeToken<List<JobExecutionPlan>>() {
+    }.getType();
+    this.serDe = new GsonSerDe<>(typeToken, serializer, deserializer);
+    this.jobExecPlanDagFactory = new JobExecutionPlanDagFactory();
+    MetricContext metricContext =
+        Instrumented.getMetricContext(new State(ConfigUtils.configToProperties(config)), this.getClass());
+    this.totalDagCount = metricContext.contextAwareCounter(ServiceMetricNames.DAG_COUNT_MYSQL_DAG_STATE_COUNT);
+
+  }
+
+  @Override
+  public void writeCheckpoint(Dag<JobExecutionPlan> dag)
+      throws IOException {
+    DagManager.DagId dagId = DagManagerUtils.generateDagId(dag);
+    boolean newDag = false;
+    for (Dag.DagNode<JobExecutionPlan> dagNode : dag.getNodes()) {
+      if (addDagNodeState(dagNode, dagId) == 1) {
+        newDag = true;
+      }
+    }
+    if (newDag) {
+      this.totalDagCount.inc();
+    }
+  }
+
+  @Override
+  public void cleanUp(Dag<JobExecutionPlan> dag) throws IOException {
+    cleanUp(generateDagId(dag).toString());

Review Comment:
   won't this call the version throwing`NotSupportedException`?



##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2Test.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import com.typesafe.config.Config;
+
+import org.apache.gobblin.config.ConfigBuilder;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.metastore.testing.ITestMetastoreDatabase;
+import org.apache.gobblin.metastore.testing.TestMetastoreDatabaseFactory;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+
+/**
+ * Mainly testing functionalities related to DagStateStore but not Mysql-related components.
+ */
+public class MysqlDagStateStoreV2Test {
+
+  private DagStateStore dagStateStore;
+
+  private static final String TEST_USER = "testUser";
+  private static ITestMetastoreDatabase testDb;
+
+  @BeforeClass
+  public void setUp() throws Exception {
+    testDb = TestMetastoreDatabaseFactory.get();
+    ConfigBuilder configBuilder = ConfigBuilder.create()
+        .addPrimitive(ConfigurationKeys.STATE_STORE_DB_USER_KEY, TEST_USER)
+        .addPrimitive(ConfigurationKeys.STATE_STORE_DB_URL_KEY, testDb.getJdbcUrl())
+        .addPrimitive(ConfigurationKeys.STATE_STORE_DB_PASSWORD_KEY, MySqlDagManagementStateStoreTest.TEST_PASSWORD);
+
+    // Constructing TopologySpecMap.
+    Map<URI, TopologySpec> topologySpecMap = new HashMap<>();
+    String specExecInstance = "mySpecExecutor";
+    TopologySpec topologySpec = DagTestUtils.buildNaiveTopologySpec(specExecInstance);
+    URI specExecURI = new URI(specExecInstance);
+    topologySpecMap.put(specExecURI, topologySpec);
+    this.dagStateStore = new MysqlDagStateStoreV2(configBuilder.build(), topologySpecMap);
+  }
+
+  @AfterClass(alwaysRun = true)
+  public void tearDown() throws Exception {
+    if (testDb != null) {
+      // `.close()` to avoid (in the aggregate, across multiple suites) - java.sql.SQLNonTransientConnectionException: Too many connections
+      testDb.close();
+    }
+  }
+
+  @Test
+  public void testWriteGetAndDeleteDag() throws Exception{
+    Dag<JobExecutionPlan> dag_0 = DagTestUtils.buildDag("random_0", 123L);
+    Dag<JobExecutionPlan> dag_1 = DagTestUtils.buildDag("random_1", 456L);

Review Comment:
   nit: having underscores or not is a funky convention; call these `dag0Orig`/`dag1Orig` vs. `dag0Retrieved`/`dag1Retrieved`



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1675436964


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStore.java:
##########


Review Comment:
   given these are controlled by config, rather than editing the class in place to transform it, why not retain the `MostlyMySqlDagManagementStateStore`?  we could always remove that later.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1680323452


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/core/GobblinServiceGuiceModule.java:
##########
@@ -178,7 +178,7 @@ public void configure(Binder binder) {
     OptionalBinder.newOptionalBinder(binder, DagActionStore.class);
     if (serviceConfig.isWarmStandbyEnabled()) {
       binder.bind(DagActionStore.class).to(MysqlDagActionStore.class);
-      binder.bind(DagManagementStateStore.class).to(MostlyMySqlDagManagementStateStore.class);
+      binder.bind(DagManagementStateStore.class).to(MySqlDagManagementStateStore.class);

Review Comment:
   in general. non-dagproc design actually just uses DMSS to access dag action stores.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1680066676


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2.java:
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagNodeStateStore} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreV2 implements DagNodeStateStore {
+
+  public static final String CONFIG_PREFIX = GOBBLIN_SERVICE_PREFIX + "mysqlDagStateStore";
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";
+  protected static final String GET_DAG_NODE_STATEMENT = "SELECT dag_node FROM %s WHERE dag_node_id = ?";
+  protected static final String DELETE_DAG_STATEMENT = "DELETE FROM %s WHERE parent_dag_id = ?";
+  protected static final String DELETE_DAG_NODE_STATEMENT = "DELETE FROM %s WHERE dag_node_id = ?";
+  private final ContextAwareCounter totalDagCount;
+
+  public MysqlDagStateStoreV2(Config config, Map<URI, TopologySpec> topologySpecMap) throws IOException {
+    if (config.hasPath(CONFIG_PREFIX)) {
+      config = config.getConfig(CONFIG_PREFIX).withFallback(config);
+    }
+
+    String DEFAULT_TABLE_NAME = "dag_node_state_store";
+    this.tableName = ConfigUtils.getString(config, ConfigurationKeys.STATE_STORE_DB_TABLE_KEY, DEFAULT_TABLE_NAME);
+    // create table if it does not exist
+    DataSource dataSource = MysqlDataSourceFactory.get(config, SharedResourcesBrokerFactory.getImplicitBroker());
+
+    try (Connection connection = dataSource.getConnection();
+        PreparedStatement createStatement = connection.prepareStatement(String.format(CREATE_TABLE_STATEMENT, tableName))) {
+      createStatement.executeUpdate();
+      connection.commit();
+    } catch (SQLException e) {
+      throw new IOException("Failure creation table " + tableName, e);
+    }
+    this.dbStatementExecutor = new DBStatementExecutor(dataSource, log);
+
+    JsonSerializer<List<JobExecutionPlan>> serializer = new JobExecutionPlanListSerializer();
+    JsonDeserializer<List<JobExecutionPlan>> deserializer = new JobExecutionPlanListDeserializer(topologySpecMap);
+    Type typeToken = new TypeToken<List<JobExecutionPlan>>() {
+    }.getType();
+    this.serDe = new GsonSerDe<>(typeToken, serializer, deserializer);
+    this.jobExecPlanDagFactory = new JobExecutionPlanDagFactory();
+    MetricContext metricContext =
+        Instrumented.getMetricContext(new State(ConfigUtils.configToProperties(config)), this.getClass());
+    this.totalDagCount = metricContext.contextAwareCounter(ServiceMetricNames.DAG_COUNT_MYSQL_DAG_STATE_COUNT);
+
+  }
+
+  @Override
+  public void writeCheckpoint(Dag<JobExecutionPlan> dag)
+      throws IOException {
+    DagManager.DagId dagId = DagManagerUtils.generateDagId(dag);
+    boolean newDag = false;
+    for (Dag.DagNode<JobExecutionPlan> dagNode : dag.getNodes()) {
+      if (addDagNodeState(dagNode, dagId) == 1) {
+        newDag = true;
+      }
+    }
+    if (newDag) {
+      this.totalDagCount.inc();
+    }
+  }
+
+  @Override
+  public void cleanUp(Dag<JobExecutionPlan> dag) throws IOException {
+    cleanUp(generateDagId(dag).toString());
+  }
+
+  @Override
+  public boolean cleanUp(DagManager.DagId dagId) throws IOException {
+    dbStatementExecutor.withPreparedStatement(String.format(DELETE_DAG_STATEMENT, tableName), deleteStatement -> {
+      try {
+        deleteStatement.setString(1, dagId.toString());

Review Comment:
   1 is the count of first `?` in the prepared statement



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1680065829


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2.java:
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagNodeStateStore} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreV2 implements DagNodeStateStore {
+
+  public static final String CONFIG_PREFIX = GOBBLIN_SERVICE_PREFIX + "mysqlDagStateStore";
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";
+  protected static final String GET_DAG_NODE_STATEMENT = "SELECT dag_node FROM %s WHERE dag_node_id = ?";
+  protected static final String DELETE_DAG_STATEMENT = "DELETE FROM %s WHERE parent_dag_id = ?";
+  protected static final String DELETE_DAG_NODE_STATEMENT = "DELETE FROM %s WHERE dag_node_id = ?";
+  private final ContextAwareCounter totalDagCount;
+
+  public MysqlDagStateStoreV2(Config config, Map<URI, TopologySpec> topologySpecMap) throws IOException {
+    if (config.hasPath(CONFIG_PREFIX)) {
+      config = config.getConfig(CONFIG_PREFIX).withFallback(config);
+    }
+
+    String DEFAULT_TABLE_NAME = "dag_node_state_store";
+    this.tableName = ConfigUtils.getString(config, ConfigurationKeys.STATE_STORE_DB_TABLE_KEY, DEFAULT_TABLE_NAME);
+    // create table if it does not exist
+    DataSource dataSource = MysqlDataSourceFactory.get(config, SharedResourcesBrokerFactory.getImplicitBroker());
+
+    try (Connection connection = dataSource.getConnection();
+        PreparedStatement createStatement = connection.prepareStatement(String.format(CREATE_TABLE_STATEMENT, tableName))) {
+      createStatement.executeUpdate();
+      connection.commit();
+    } catch (SQLException e) {
+      throw new IOException("Failure creation table " + tableName, e);
+    }
+    this.dbStatementExecutor = new DBStatementExecutor(dataSource, log);
+
+    JsonSerializer<List<JobExecutionPlan>> serializer = new JobExecutionPlanListSerializer();
+    JsonDeserializer<List<JobExecutionPlan>> deserializer = new JobExecutionPlanListDeserializer(topologySpecMap);
+    Type typeToken = new TypeToken<List<JobExecutionPlan>>() {
+    }.getType();
+    this.serDe = new GsonSerDe<>(typeToken, serializer, deserializer);
+    this.jobExecPlanDagFactory = new JobExecutionPlanDagFactory();
+    MetricContext metricContext =
+        Instrumented.getMetricContext(new State(ConfigUtils.configToProperties(config)), this.getClass());
+    this.totalDagCount = metricContext.contextAwareCounter(ServiceMetricNames.DAG_COUNT_MYSQL_DAG_STATE_COUNT);
+
+  }
+
+  @Override
+  public void writeCheckpoint(Dag<JobExecutionPlan> dag)
+      throws IOException {
+    DagManager.DagId dagId = DagManagerUtils.generateDagId(dag);
+    boolean newDag = false;
+    for (Dag.DagNode<JobExecutionPlan> dagNode : dag.getNodes()) {
+      if (addDagNodeState(dagNode, dagId) == 1) {
+        newDag = true;
+      }
+    }
+    if (newDag) {
+      this.totalDagCount.inc();
+    }
+  }
+
+  @Override
+  public void cleanUp(Dag<JobExecutionPlan> dag) throws IOException {
+    cleanUp(generateDagId(dag).toString());

Review Comment:
   corrected



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1681584932


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStore.java:
##########


Review Comment:
   We cannot use `MostlyMySqlDagManagementStateStore` in distributed mode, with or without dag proc engine. So it would not be a good idea to move back to MostlyMySqlDagManagementStateStore if anything does not work.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1680013089


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStore.java:
##########
@@ -62,13 +59,10 @@
  */
 @Slf4j
 @Singleton
-public class MostlyMySqlDagManagementStateStore implements DagManagementStateStore {
-  private final Map<DagNodeId, Dag.DagNode<JobExecutionPlan>> dagNodes = new ConcurrentHashMap<>();
-  // dagToJobs holds a map of dagId to running jobs of that dag
-  private final Map<DagManager.DagId, Set<Dag.DagNode<JobExecutionPlan>>> dagToJobs = new ConcurrentHashMap<>();
-  private DagStateStore dagStateStore;
-  private DagStateStore failedDagStateStore;
-  private JobStatusRetriever jobStatusRetriever;
+public class MySqlDagManagementStateStore implements DagManagementStateStore {

Review Comment:
   but if it uses MysqlDagStateStore it has to use non-mysql data structure (maps) for proper functioning 



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1680004822


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStore.java:
##########
@@ -62,13 +59,10 @@
  */
 @Slf4j
 @Singleton
-public class MostlyMySqlDagManagementStateStore implements DagManagementStateStore {
-  private final Map<DagNodeId, Dag.DagNode<JobExecutionPlan>> dagNodes = new ConcurrentHashMap<>();
-  // dagToJobs holds a map of dagId to running jobs of that dag
-  private final Map<DagManager.DagId, Set<Dag.DagNode<JobExecutionPlan>>> dagToJobs = new ConcurrentHashMap<>();
-  private DagStateStore dagStateStore;
-  private DagStateStore failedDagStateStore;
-  private JobStatusRetriever jobStatusRetriever;
+public class MySqlDagManagementStateStore implements DagManagementStateStore {

Review Comment:
   I mean, I used that case because it is actually written `MySQL`



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2115] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1683386592


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ReevaluateDagProc.java:
##########
@@ -167,9 +160,5 @@ private void onJobFinish(DagManagementStateStore dagManagementStateStore, Dag.Da
       default:
         log.warn("It should not reach here. Job status {} is unexpected.", executionStatus);
     }
-
-    // Checkpoint the dag state, it should have an updated value of dag fields
-    dagManagementStateStore.checkpointDag(dag);

Review Comment:
   yes. only setMessage and setFlowEvent are called, which are anyway not persisted anyway



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1680025630


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2.java:
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagNodeStateStore} using MySQL as a backup.

Review Comment:
   actually renamed it to MysqlDagStateStoreWithDagNodes. new name also emphasize on the difference. this is a `DagStateStoreWithDagNodes` while MysqlDagStateStore is a `DagStateStore`



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1681810288


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/EnforceFlowFinishDeadlineDagProc.java:
##########
@@ -58,7 +58,6 @@ protected void enforceDeadline(DagManagementStateStore dagManagementStateStore,
 
       dag.setFlowEvent(TimingEvent.FlowTimings.FLOW_RUN_DEADLINE_EXCEEDED);
       dag.setMessage("Flow killed due to exceeding SLA of " + flowFinishDeadline + " ms");
-      dagManagementStateStore.checkpointDag(dag);

Review Comment:
   i was "checkpointing" them because i was changing "flowEvent" and "message" fields. but i realize these two fields are not persisted anyway.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2115] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1683380540


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreWithDagNodes.java:
##########
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagStateStoreWithDagNodes} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreWithDagNodes implements DagStateStoreWithDagNodes {
+
+  public static final String CONFIG_PREFIX = MysqlDagStateStore.CONFIG_PREFIX;
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON NOT NULL, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id), "
+      + "UNIQUE INDEX dag_node_index (dag_node_id), "
+      + "INDEX dag_index (parent_dag_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_NODES_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";
+  protected static final String GET_DAG_NODE_STATEMENT = "SELECT dag_node FROM %s WHERE dag_node_id = ?";
+  protected static final String DELETE_DAG_STATEMENT = "DELETE FROM %s WHERE parent_dag_id = ?";
+  private final ContextAwareCounter totalDagCount;
+
+  public MysqlDagStateStoreWithDagNodes(Config config, Map<URI, TopologySpec> topologySpecMap) throws IOException {
+    if (config.hasPath(CONFIG_PREFIX)) {
+      config = config.getConfig(CONFIG_PREFIX).withFallback(config);
+    }
+
+    String DEFAULT_TABLE_NAME = "dag_node_state_store";
+    this.tableName = ConfigUtils.getString(config, ConfigurationKeys.STATE_STORE_DB_TABLE_KEY, DEFAULT_TABLE_NAME);
+    // create table if it does not exist
+    DataSource dataSource = MysqlDataSourceFactory.get(config, SharedResourcesBrokerFactory.getImplicitBroker());
+
+    try (Connection connection = dataSource.getConnection();
+        PreparedStatement createStatement = connection.prepareStatement(String.format(CREATE_TABLE_STATEMENT, tableName))) {
+      createStatement.executeUpdate();
+      connection.commit();
+    } catch (SQLException e) {
+      throw new IOException("Failure creation table " + tableName, e);
+    }
+    this.dbStatementExecutor = new DBStatementExecutor(dataSource, log);
+
+    JsonSerializer<List<JobExecutionPlan>> serializer = new JobExecutionPlanListSerializer();
+    JsonDeserializer<List<JobExecutionPlan>> deserializer = new JobExecutionPlanListDeserializer(topologySpecMap);
+    Type typeToken = new TypeToken<List<JobExecutionPlan>>() {
+    }.getType();
+    this.serDe = new GsonSerDe<>(typeToken, serializer, deserializer);
+    this.jobExecPlanDagFactory = new JobExecutionPlanDagFactory();
+    MetricContext metricContext =
+        Instrumented.getMetricContext(new State(ConfigUtils.configToProperties(config)), this.getClass());
+    this.totalDagCount = metricContext.contextAwareCounter(ServiceMetricNames.DAG_COUNT_MYSQL_DAG_STATE_COUNT);
+    log.info("Instantiated {}", getClass().getSimpleName());
+  }
+
+  @Override
+  public void writeCheckpoint(Dag<JobExecutionPlan> dag)

Review Comment:
   my suggestion is to rename this to `addDag`.
   update is for dag nodes



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2115] implement DagNodeStateStore [gobblin]

Posted by "Will-Lo (via GitHub)" <gi...@apache.org>.
Will-Lo commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1683416757


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStore.java:
##########
@@ -116,128 +112,78 @@ public void removeFlowSpec(URI uri, Properties headers, boolean triggerListener)
     this.flowCatalog.remove(uri, headers, triggerListener);
   }
 
-  public synchronized void setTopologySpecMap(Map<URI, TopologySpec> topologySpecMap) throws IOException {
+  public synchronized void setTopologySpecMap(Map<URI, TopologySpec> topologySpecMap) {
     this.topologySpecMap = topologySpecMap;
     start();
   }
 
-  private DagStateStore createDagStateStore(Config config, Map<URI, TopologySpec> topologySpecMap) {
+  private DagStateStoreWithDagNodes createDagStateStore(Config config, Map<URI, TopologySpec> topologySpecMap) {
     try {
-      Class<?> dagStateStoreClass = Class.forName(ConfigUtils.getString(config, DAG_STATESTORE_CLASS_KEY, MysqlDagStateStore.class.getName()));
-      return (DagStateStore) GobblinConstructorUtils.invokeLongestConstructor(dagStateStoreClass, config, topologySpecMap);
+      Class<?> dagStateStoreClass = Class.forName(ConfigUtils.getString(config, DAG_STATESTORE_CLASS_KEY, MysqlDagStateStoreWithDagNodes.class.getName()));
+      return (DagStateStoreWithDagNodes) GobblinConstructorUtils.invokeLongestConstructor(dagStateStoreClass, config, topologySpecMap);
     } catch (ReflectiveOperationException e) {
       throw new RuntimeException(e);
     }
   }
 
   @Override
-  public void checkpointDag(Dag<JobExecutionPlan> dag) throws IOException {

Review Comment:
   I think this makes sense primarily if the dag is immutable (i.e. you are certain when calling this line that you are creating a new row and adding a dag that already exists may result in an error) otherwise checkpoint may still make sense here semantically.



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementStateStore.java:
##########
@@ -66,25 +65,11 @@ public interface DagManagementStateStore {
    */
   void checkpointDag(Dag<JobExecutionPlan> dag) throws IOException;
 
-  /**
-   @return whether `dagId` is currently known due to {@link DagManagementStateStore#checkpointDag} but not yet
-   {@link DagManagementStateStore#deleteDag}
-   */
-  boolean containsDag(DagManager.DagId dagId) throws IOException;
-
   /**
    @return the {@link Dag}, if present
    */
   Optional<Dag<JobExecutionPlan>> getDag(DagManager.DagId dagId) throws IOException;
 
-  /**
-   * Delete the {@link Dag} from the backing store, typically called upon completion of execution.
-   * @param dag The dag completed/cancelled execution on {@link org.apache.gobblin.runtime.api.SpecExecutor}.
-   */
-  default void deleteDag(Dag<JobExecutionPlan> dag) throws IOException {
-    deleteDag(DagManagerUtils.generateDagId(dag));
-  }
-

Review Comment:
   IMO it comes down to how the callers perform work. Do they only store IDs and this implementation of the state store does all the mutation and modification of the dags? If so, then we should have a consistent API that takes in DagIds as the function parameters. If the callers themselves are modifying dags, and this class is only used as a storage mechanism, may make sense to just accept a full dag as the parameter for better encapsulation.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2115] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1683445002


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagActionStore.java:
##########
@@ -49,26 +48,22 @@ public class MysqlDagActionStore implements DagActionStore {
   protected final DataSource dataSource;
   private final DBStatementExecutor dbStatementExecutor;
   private final String tableName;
-  private final long retentionPeriodSeconds;
-  private String thisTableRetentionStatement;
   private static final String EXISTS_STATEMENT = "SELECT EXISTS(SELECT * FROM %s WHERE flow_group = ? AND flow_name = ? AND flow_execution_id = ? AND job_name = ? AND dag_action = ?)";
 
   protected static final String INSERT_STATEMENT = "INSERT INTO %s (flow_group, flow_name, flow_execution_id, job_name, dag_action) "
       + "VALUES (?, ?, ?, ?, ?)";
   private static final String DELETE_STATEMENT = "DELETE FROM %s WHERE flow_group = ? AND flow_name =? AND flow_execution_id = ? AND job_name = ? AND dag_action = ?";
-  private static final String GET_STATEMENT = "SELECT flow_group, flow_name, flow_execution_id, job_name, dag_action FROM %s WHERE flow_group = ? AND flow_name =? AND flow_execution_id = ? AND job_name = ? AND dag_action = ?";

Review Comment:
   no one dag action is never desired... only requirement is to load all the dag actions in the beginning



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagActionStore.java:
##########
@@ -49,26 +48,22 @@ public class MysqlDagActionStore implements DagActionStore {
   protected final DataSource dataSource;
   private final DBStatementExecutor dbStatementExecutor;
   private final String tableName;
-  private final long retentionPeriodSeconds;
-  private String thisTableRetentionStatement;
   private static final String EXISTS_STATEMENT = "SELECT EXISTS(SELECT * FROM %s WHERE flow_group = ? AND flow_name = ? AND flow_execution_id = ? AND job_name = ? AND dag_action = ?)";
 
   protected static final String INSERT_STATEMENT = "INSERT INTO %s (flow_group, flow_name, flow_execution_id, job_name, dag_action) "
       + "VALUES (?, ?, ?, ?, ?)";
   private static final String DELETE_STATEMENT = "DELETE FROM %s WHERE flow_group = ? AND flow_name =? AND flow_execution_id = ? AND job_name = ? AND dag_action = ?";
-  private static final String GET_STATEMENT = "SELECT flow_group, flow_name, flow_execution_id, job_name, dag_action FROM %s WHERE flow_group = ? AND flow_name =? AND flow_execution_id = ? AND job_name = ? AND dag_action = ?";

Review Comment:
   no, one dag action is never desired... only requirement is to load all the dag actions in the beginning



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2115] implement DagNodeStateStore [gobblin]

Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1683438110


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagStateStoreWithDagNodes.java:
##########
@@ -33,9 +33,22 @@
  * {@link DagStateStore#cleanUp(DagManager.DagId)} should be used to delete all the {@link Dag.DagNode}s for a {@link Dag}.
  */
 public interface DagStateStoreWithDagNodes extends DagStateStore {
+
+  /**
+   * updates a dag node identified by the provided {@link org.apache.gobblin.service.modules.orchestration.DagManager.DagId}

Review Comment:
   nit can you capitalize the first word of these docs. Explain what the int it returns represents



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1680059863


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreV2.java:
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagNodeStateStore} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreV2 implements DagNodeStateStore {
+
+  public static final String CONFIG_PREFIX = GOBBLIN_SERVICE_PREFIX + "mysqlDagStateStore";
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";
+  protected static final String GET_DAG_NODE_STATEMENT = "SELECT dag_node FROM %s WHERE dag_node_id = ?";
+  protected static final String DELETE_DAG_STATEMENT = "DELETE FROM %s WHERE parent_dag_id = ?";
+  protected static final String DELETE_DAG_NODE_STATEMENT = "DELETE FROM %s WHERE dag_node_id = ?";
+  private final ContextAwareCounter totalDagCount;
+
+  public MysqlDagStateStoreV2(Config config, Map<URI, TopologySpec> topologySpecMap) throws IOException {
+    if (config.hasPath(CONFIG_PREFIX)) {
+      config = config.getConfig(CONFIG_PREFIX).withFallback(config);
+    }
+
+    String DEFAULT_TABLE_NAME = "dag_node_state_store";

Review Comment:
   but it can be local



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1680122092


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -90,7 +90,7 @@ protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag
     dagManagementStateStore.checkpointDag(failedDag.get());
 
     // if it fails here, it will check point the failed dag in the (running) dag store again, which is idempotent
-    dagManagementStateStore.deleteFailedDag(failedDag.get());
+    dagManagementStateStore.deleteFailedDag(getDagId());

Review Comment:
   I think we should expect this kind of symmetry. It is not like a rest resource, where we do `get resource` and `delete resource`.
   It is a table with <dagId, dag>. Consider it like a `Map<K, V>` with auto incrementing key, we do not call `delete<V>` just because we do `add(V)`.
   There are many other examples like, `Stack` `Queue` API. 
   Basically, I believe, a `key` (dagId/dagNodeId) is the best way to choose a resource/row/dag i**n this case**, not the dag.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2115] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1683358678


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementStateStore.java:
##########
@@ -42,6 +41,8 @@
  * 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.
  */
+// todo - this should merge with DagStateStoreWithDagNodes interface and `addDagNodeState` and `addDagNode` should merge and rename to

Review Comment:
   they are separate entities. `DagManagementStateStore` encompasses all the things we persist - DagStates/DagNodeStates, DagActions, UserQuotas, FlowSpecs



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2115] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1683383775


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/DagProcUtils.java:
##########
@@ -169,8 +169,8 @@ public static void cancelDagNode(Dag.DagNode<JobExecutionPlan> dagNodeToCancel,
             dagNodeToCancel.getValue().getJobSpec().getUri());
       }
       DagManagerUtils.getSpecProducer(dagNodeToCancel).cancelJob(dagNodeToCancel.getValue().getJobSpec().getUri(), props).get();
-      // todo - why was it not being cleaned up in DagManager?
-      dagManagementStateStore.deleteDagNodeState(dagId, dagNodeToCancel);
+      // add back the dag node with updated states in the store
+      dagManagementStateStore.addDagNodeState(dagNodeToCancel, dagId);

Review Comment:
   right



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2115] implement DagNodeStateStore [gobblin]

Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1683434920


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreWithDagNodes.java:
##########
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.eclipse.jgit.errors.NotSupportedException;
+
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import javax.sql.DataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.broker.SharedResourcesBrokerFactory;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metastore.MysqlDataSourceFactory;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.spec_serde.GsonSerDe;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanDagFactory;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListDeserializer;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlanListSerializer;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.DBStatementExecutor;
+
+import static org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateDagId;
+
+
+/**
+ * An implementation of {@link DagStateStoreWithDagNodes} using MySQL as a backup.
+ */
+@Slf4j
+public class MysqlDagStateStoreWithDagNodes implements DagStateStoreWithDagNodes {
+
+  public static final String CONFIG_PREFIX = MysqlDagStateStore.CONFIG_PREFIX;
+  protected final DBStatementExecutor dbStatementExecutor;
+  protected final String tableName;
+  protected final GsonSerDe<List<JobExecutionPlan>> serDe;
+  private final JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+  // todo add a column that tells if it is a running dag or a failed dag
+  protected static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+      + "dag_node_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_NODE_ID_LENGTH + ") CHARACTER SET latin1 COLLATE latin1_bin NOT NULL, "
+      + "parent_dag_id VARCHAR(" + ServiceConfigKeys.MAX_DAG_ID_LENGTH + ") NOT NULL, "
+      + "dag_node JSON NOT NULL, "
+      + "modified_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, "
+      + "PRIMARY KEY (dag_node_id), "
+      + "UNIQUE INDEX dag_node_index (dag_node_id), "
+      + "INDEX dag_index (parent_dag_id))";
+
+  protected static final String INSERT_STATEMENT = "INSERT INTO %s (dag_node_id, parent_dag_id, dag_node) "
+      + "VALUES (?, ?, ?) AS new ON DUPLICATE KEY UPDATE dag_node = new.dag_node";
+  protected static final String GET_DAG_NODES_STATEMENT = "SELECT dag_node FROM %s WHERE parent_dag_id = ?";
+  protected static final String GET_DAG_NODE_STATEMENT = "SELECT dag_node FROM %s WHERE dag_node_id = ?";
+  protected static final String DELETE_DAG_STATEMENT = "DELETE FROM %s WHERE parent_dag_id = ?";
+  private final ContextAwareCounter totalDagCount;
+
+  public MysqlDagStateStoreWithDagNodes(Config config, Map<URI, TopologySpec> topologySpecMap) throws IOException {
+    if (config.hasPath(CONFIG_PREFIX)) {
+      config = config.getConfig(CONFIG_PREFIX).withFallback(config);
+    }
+
+    String DEFAULT_TABLE_NAME = "dag_node_state_store";
+    this.tableName = ConfigUtils.getString(config, ConfigurationKeys.STATE_STORE_DB_TABLE_KEY, DEFAULT_TABLE_NAME);
+    // create table if it does not exist
+    DataSource dataSource = MysqlDataSourceFactory.get(config, SharedResourcesBrokerFactory.getImplicitBroker());
+
+    try (Connection connection = dataSource.getConnection();
+        PreparedStatement createStatement = connection.prepareStatement(String.format(CREATE_TABLE_STATEMENT, tableName))) {
+      createStatement.executeUpdate();
+      connection.commit();
+    } catch (SQLException e) {
+      throw new IOException("Failure creation table " + tableName, e);
+    }
+    this.dbStatementExecutor = new DBStatementExecutor(dataSource, log);
+
+    JsonSerializer<List<JobExecutionPlan>> serializer = new JobExecutionPlanListSerializer();
+    JsonDeserializer<List<JobExecutionPlan>> deserializer = new JobExecutionPlanListDeserializer(topologySpecMap);
+    Type typeToken = new TypeToken<List<JobExecutionPlan>>() {
+    }.getType();
+    this.serDe = new GsonSerDe<>(typeToken, serializer, deserializer);
+    this.jobExecPlanDagFactory = new JobExecutionPlanDagFactory();
+    MetricContext metricContext =
+        Instrumented.getMetricContext(new State(ConfigUtils.configToProperties(config)), this.getClass());
+    this.totalDagCount = metricContext.contextAwareCounter(ServiceMetricNames.DAG_COUNT_MYSQL_DAG_STATE_COUNT);
+    log.info("Instantiated {}", getClass().getSimpleName());
+  }
+
+  @Override
+  public void writeCheckpoint(Dag<JobExecutionPlan> dag)

Review Comment:
   yes let's rename to `addDag`



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2115] implement DagNodeStateStore [gobblin]

Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1683486220


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementStateStore.java:
##########
@@ -42,6 +41,8 @@
  * 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.
  */
+// todo - this should merge with DagStateStoreWithDagNodes interface and `addDagNodeState` and `addDagNode` should merge and rename to

Review Comment:
   Realized this after reading further along. I got confused by the names. DMSS encompasses all.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1680122092


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -90,7 +90,7 @@ protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag
     dagManagementStateStore.checkpointDag(failedDag.get());
 
     // if it fails here, it will check point the failed dag in the (running) dag store again, which is idempotent
-    dagManagementStateStore.deleteFailedDag(failedDag.get());
+    dagManagementStateStore.deleteFailedDag(getDagId());

Review Comment:
   I think we should expect this kind of symmetry. It is not like a rest resource, where we do `get resource` and `delete resource`.
   It is a table with <dagId, dag>. Consider it like a `Map<K, V>` with auto incrementing key, we do not call `delete<V>` just because we do `add(V)`.
   There are many other examples like, `Stack` `Queue` API. 
   Basically, I believe, a `key` (dagId/dagNodeId) is the best way to choose a resource/row/dag **in this case**, not the dag.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1680122092


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -90,7 +90,7 @@ protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag
     dagManagementStateStore.checkpointDag(failedDag.get());
 
     // if it fails here, it will check point the failed dag in the (running) dag store again, which is idempotent
-    dagManagementStateStore.deleteFailedDag(failedDag.get());
+    dagManagementStateStore.deleteFailedDag(getDagId());

Review Comment:
   I think we should expect this kind of symmetry. It is not like a rest resource, where we do `get resource` and `delete resource`.
   It is a table with <dagId, dag>. Consider it like a `Map<K, V>` with auto incrementing key, we do not call `delete<V>` just because we do `add(V)`.
   There are many other examples like, `Stack` `Queue` API. 
   Basically, I believe, a `key` (dagId/dagNodeId) is a best way to choose a resource/row/dag in this case, not the dag.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1675031420


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagStateStore.java:
##########
@@ -47,28 +47,42 @@ public interface DagStateStore {
    */
   void cleanUp(Dag<JobExecutionPlan> dag) throws IOException;
 
+  default boolean cleanUp(DagManager.DagId dagId) throws IOException {
+    cleanUp(dagId.toString());
+    return true;
+  }
+
   /**
    * Delete the {@link Dag} from the backing store, typically upon completion of execution.
    * @param dagId The ID of the dag to clean up.
    */
+  // todo - change it to return boolean
+  @Deprecated

Review Comment:
   keeping them because DagManager uses it



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1681836322


##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStoreTest.java:
##########
@@ -90,41 +103,40 @@ public void testAddDag() throws Exception {
     DagManager.DagId dagId2 = DagManagerUtils.generateDagId(dag2);
     DagNodeId dagNodeId = DagManagerUtils.calcJobId(dagNode.getValue().getJobSpec().getConfig());
 
-    this.dagManagementStateStore.checkpointDag(dag);
-    this.dagManagementStateStore.checkpointDag(dag2);
+    this.dagManagementStateStore.addDag(dag);
+    this.dagManagementStateStore.addDag(dag2);
     this.dagManagementStateStore.addDagNodeState(dagNode, dagId);
     this.dagManagementStateStore.addDagNodeState(dagNode2, dagId);
     this.dagManagementStateStore.addDagNodeState(dagNode3, dagId2);
 
-    Assert.assertTrue(this.dagManagementStateStore.containsDag(dagId));
-    Assert.assertEquals(dag.toString(), this.dagManagementStateStore.getDag(dagId).get().toString());
+    Assert.assertTrue(compareLists(dag.getNodes(), this.dagManagementStateStore.getDag(dagId).get().getNodes()));
     Assert.assertEquals(dagNode, this.dagManagementStateStore.getDagNodeWithJobStatus(dagNodeId).getLeft().get());
 
     Set<Dag.DagNode<JobExecutionPlan>> dagNodes = this.dagManagementStateStore.getDagNodes(dagId);
     Assert.assertEquals(2, dagNodes.size());
     Assert.assertTrue(dagNodes.contains(dagNode));
     Assert.assertTrue(dagNodes.contains(dagNode2));
 
-    this.dagManagementStateStore.deleteDagNodeState(dagId, dagNode);

Review Comment:
   there is no delete dag node happen now



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2115] implement DagNodeStateStore [gobblin]

Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1683439661


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagActionStore.java:
##########
@@ -49,26 +48,22 @@ public class MysqlDagActionStore implements DagActionStore {
   protected final DataSource dataSource;
   private final DBStatementExecutor dbStatementExecutor;
   private final String tableName;
-  private final long retentionPeriodSeconds;
-  private String thisTableRetentionStatement;
   private static final String EXISTS_STATEMENT = "SELECT EXISTS(SELECT * FROM %s WHERE flow_group = ? AND flow_name = ? AND flow_execution_id = ? AND job_name = ? AND dag_action = ?)";
 
   protected static final String INSERT_STATEMENT = "INSERT INTO %s (flow_group, flow_name, flow_execution_id, job_name, dag_action) "
       + "VALUES (?, ?, ?, ?, ?)";
   private static final String DELETE_STATEMENT = "DELETE FROM %s WHERE flow_group = ? AND flow_name =? AND flow_execution_id = ? AND job_name = ? AND dag_action = ?";
-  private static final String GET_STATEMENT = "SELECT flow_group, flow_name, flow_execution_id, job_name, dag_action FROM %s WHERE flow_group = ? AND flow_name =? AND flow_execution_id = ? AND job_name = ? AND dag_action = ?";

Review Comment:
   is this never used



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2115] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1683378178


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStore.java:
##########
@@ -62,13 +60,11 @@
  */
 @Slf4j
 @Singleton
-public class MostlyMySqlDagManagementStateStore implements DagManagementStateStore {

Review Comment:
   this is not that kinda store, it is like a group of different stores, each of which has those sql statements



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2115] implement DagNodeStateStore [gobblin]

Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3999:
URL: https://github.com/apache/gobblin/pull/3999#discussion_r1683388463


##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStoreTest.java:
##########
@@ -90,41 +103,40 @@ public void testAddDag() throws Exception {
     DagManager.DagId dagId2 = DagManagerUtils.generateDagId(dag2);
     DagNodeId dagNodeId = DagManagerUtils.calcJobId(dagNode.getValue().getJobSpec().getConfig());
 
-    this.dagManagementStateStore.checkpointDag(dag);
-    this.dagManagementStateStore.checkpointDag(dag2);
+    this.dagManagementStateStore.addDag(dag);
+    this.dagManagementStateStore.addDag(dag2);
     this.dagManagementStateStore.addDagNodeState(dagNode, dagId);
     this.dagManagementStateStore.addDagNodeState(dagNode2, dagId);
     this.dagManagementStateStore.addDagNodeState(dagNode3, dagId2);
 
-    Assert.assertTrue(this.dagManagementStateStore.containsDag(dagId));
-    Assert.assertEquals(dag.toString(), this.dagManagementStateStore.getDag(dagId).get().toString());
+    Assert.assertTrue(compareLists(dag.getNodes(), this.dagManagementStateStore.getDag(dagId).get().getNodes()));
     Assert.assertEquals(dagNode, this.dagManagementStateStore.getDagNodeWithJobStatus(dagNodeId).getLeft().get());
 
     Set<Dag.DagNode<JobExecutionPlan>> dagNodes = this.dagManagementStateStore.getDagNodes(dagId);
     Assert.assertEquals(2, dagNodes.size());
     Assert.assertTrue(dagNodes.contains(dagNode));
     Assert.assertTrue(dagNodes.contains(dagNode2));
 
-    this.dagManagementStateStore.deleteDagNodeState(dagId, dagNode);
-    Assert.assertFalse(this.dagManagementStateStore.getDagNodes(dagId).contains(dagNode));
-    Assert.assertTrue(this.dagManagementStateStore.getDagNodes(dagId).contains(dagNode2));
-    Assert.assertTrue(this.dagManagementStateStore.getDagNodes(dagId2).contains(dagNode3));
-
     // test to verify that adding a new dag node with the same dag node id (defined by the jobSpec) replaces the existing one
-    Assert.assertEquals(this.dagManagementStateStore.getDagNodes(dagId).size(), 1);

Review Comment:
   we did that right above.... somewhat hidden because of the diffs in github UI



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org