[GOBBLIN-2137]merged dagNodeStateStore and failedDagNodeStateStore tables #4032

pratapaditya04 · 2024-08-16T09:31:01Z

Dear Gobblin maintainers,

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

JIRA

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

Description

Here are some details about my PR, including screenshots (if applicable):
Right now we are maintaining two tables to maintain DagState and Failed Dag State, In this PR ,we have tried to merge FailedDagState tables into DagState by adding a column is_failed_dag in DagState

Tests

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

Commits

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

codecov-commenter · 2024-08-19T21:11:59Z

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 38.80%. Comparing base (e501b62) to head (7e3e556).
Report is 9 commits behind head on master.

❗ There is a different number of reports uploaded between BASE (e501b62) and HEAD (7e3e556). Click for more details.

HEAD has 1 upload less than BASE

Flag BASE (e501b62) HEAD (7e3e556)

2 1

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #4032      +/-   ##
============================================
- Coverage     45.86%   38.80%   -7.06%     
+ Complexity     3257     1599    -1658     
============================================
  Files           707      388     -319     
  Lines         27865    15995   -11870     
  Branches       2796     1585    -1211     
============================================
- Hits          12779     6207    -6572     
+ Misses        14008     9290    -4718     
+ Partials       1078      498     -580

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

…nto failed_dag_store_table_merger # Conflicts: # gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/Dag.java # gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreWithDagNodes.java # gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreWithDagNodesTest.java

arjun4084346 · 2024-08-20T17:52:49Z

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

@@ -48,6 +48,8 @@ public class Dag<T> {
  // Map to maintain parent to children mapping.
  private Map<DagNode, List<DagNode<T>>> parentChildMap;
  private List<DagNode<T>> nodes;
+  @Setter


because we do not persist dag level field in mysql, adding fields to Dag will not be much useful and may lead to bugs

We want to have this field here as we didn't want to add additional parameters in all the methods to pass on is_failed value.

...rc/main/java/org/apache/gobblin/service/modules/orchestration/DagStateStoreWithDagNodes.java

...in/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreWithDagNodes.java

arjun4084346 · 2024-08-20T19:05:55Z

Should
a) isFailedDag be ONLY within DagNode always, everywhere, with up-to-date value?
b) isFailedDag be within DagNode, but be also in mysql table as a cache; still be always in sync with the field inside DagNode? It may make sql queries more readable.

before completing the review, i would like to understand your thoughts on this.

pratapaditya04 · 2024-08-21T15:21:20Z

Should a) isFailedDag be ONLY within DagNode always, everywhere, with up-to-date value? b) isFailedDag be within DagNode, but be also in mysql table as a cache; still be always in sync with the field inside DagNode? It may make sql queries more readable.

before completing the review, i would like to understand your thoughts on this.

If we don't keep it in mysql, we may lose it in case of restarts/deployments, so we will have to store in mysql

...in/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreWithDagNodes.java

phet

this will be a nice simplification. looks close

phet · 2024-09-25T22:23:11Z

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

@@ -77,22 +77,11 @@ public interface DagManagementStateStore {

  /**
   * This marks the dag as a failed one.
-   * Failed dags are queried using {@link DagManagementStateStore#getFailedDag(DagManager.DagId)} ()} later to be retried.
+   * Failed dags are queried using {@link DagManagementStateStore#getDag(DagManager.DagId)} ()} later to be retried.


does it remain useful to both retrieve the DAG while also asserting that it's failed?

...rc/main/java/org/apache/gobblin/service/modules/orchestration/DagStateStoreWithDagNodes.java

phet · 2024-09-25T22:40:09Z

...main/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStore.java

@@ -91,8 +88,6 @@ public MySqlDagManagementStateStore(Config config, FlowCatalog flowCatalog, User
  private synchronized void start() {
    if (!dagStoresInitialized) {
      this.dagStateStore = createDagStateStore(config, topologySpecMap);
-      this.failedDagStateStore = createDagStateStore(ConfigUtils.getConfigOrEmpty(config, FAILED_DAG_STATESTORE_PREFIX).withFallback(config),


any ideas on handling migration when we roll this out (presuming the failed DagStateStore was not empty)?

phet · 2024-09-25T22:43:28Z

...main/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStore.java

  @Override
  public synchronized void addDagNodeState(Dag.DagNode<JobExecutionPlan> dagNode, DagManager.DagId dagId)
      throws IOException {
-    this.dagStateStore.updateDagNode(dagId, dagNode);
+    this.dagStateStore.updateDagNode(dagId, dagNode, false);// isFailedDag is set as false because addDagNodeState adds a new DagNode, doesn't update an existing dagNode as failed.


nit: space before starting a comment. also more brevity; e.g.:

// create all DagNodes as isFailedDag == false

phet · 2024-09-25T22:47:07Z

...in/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreWithDagNodes.java

+        PreparedStatement createStatement = connection.prepareStatement(
+            String.format(CREATE_TABLE_STATEMENT, tableName))) {


given arjun just wrote this class a month or two back, please ensure your auto-formatting is what it's supposed to be. it is possible his was off, but let's check. sure we might fix spelling errors, but there should be little reason to reformat files we've only just created

phet · 2024-09-25T22:48:32Z

...in/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreWithDagNodes.java

@@ -167,7 +167,7 @@ public void cleanUp(String dagId) throws IOException {
  @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");  }
+        + "the DagManager that is replaced by DagProcessingEngine");}


actually, doesn't this need a newline before }?

phet · 2024-09-25T22:50:06Z

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

@@ -54,7 +54,7 @@ public ResumeDagProc(ResumeDagTask resumeDagTask, Config config) {
  @Override
  protected Optional<Dag<JobExecutionPlan>> initialize(DagManagementStateStore dagManagementStateStore)
      throws IOException {
-      return dagManagementStateStore.getFailedDag(getDagId());
+      return dagManagementStateStore.getDag(getDagId());


shall we verify the one returned is actually failed?

phet · 2024-09-25T22:53:55Z

...ava/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreWithDagNodesTest.java

+  private List<Boolean> fetchDagNodeStates(String dagId) throws IOException {
+    List<Boolean> states = new ArrayList<>();
+
+    dbStatementExecutor.withPreparedStatement(String.format(GET_DAG_NODES_STATEMENT, tableName), getStatement -> {


is behind-the-scenes DB access the only way to validate behavior here? is there no way to access from the "official" DagStateStore, then mark failed and finally re-access from the DSS to verify all nodes have changed?

merged dagNodeStateStore and failedDagNodeStateStore tables

719052f

pratapaditya04 force-pushed the failed_dag_store_table_merger branch from e08ac21 to 719052f Compare August 19, 2024 06:51

merged dagNodeStateStore and failedDagNodeStateStore tables

68f6ea2

Aditya Pratap Singh added 4 commits August 20, 2024 12:04

added isFailedDag at Node level

8614a0d

cleaned up failed dag state store references

dcf6e1a

refactored

7883009

pratapaditya04 changed the title ~~[WIP]merged dagNodeStateStore and failedDagNodeStateStore tables~~ [GOBBLIN-2137]merged dagNodeStateStore and failedDagNodeStateStore tables Aug 20, 2024

arjun4084346 reviewed Aug 20, 2024

View reviewed changes

...rc/main/java/org/apache/gobblin/service/modules/orchestration/DagStateStoreWithDagNodes.java Show resolved Hide resolved

arjun4084346 reviewed Aug 20, 2024

View reviewed changes

...in/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreWithDagNodes.java Outdated Show resolved Hide resolved

arjun4084346 reviewed Aug 20, 2024

View reviewed changes

...in/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreWithDagNodes.java Outdated Show resolved Hide resolved

addressed PR comments

7e3e556

pratapaditya04 force-pushed the failed_dag_store_table_merger branch from c13315d to 7e3e556 Compare August 21, 2024 15:20

abhishekmjain reviewed Aug 22, 2024

View reviewed changes

...in/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStoreWithDagNodes.java Show resolved Hide resolved

phet reviewed Sep 25, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[GOBBLIN-2137]merged dagNodeStateStore and failedDagNodeStateStore tables #4032

[GOBBLIN-2137]merged dagNodeStateStore and failedDagNodeStateStore tables #4032

pratapaditya04 commented Aug 16, 2024 •

edited

Loading

codecov-commenter commented Aug 19, 2024 •

edited

Loading

arjun4084346 Aug 20, 2024

pratapaditya04 Aug 21, 2024

arjun4084346 commented Aug 20, 2024

pratapaditya04 commented Aug 21, 2024

phet left a comment

phet Sep 25, 2024

phet Sep 25, 2024

phet Sep 25, 2024

phet Sep 25, 2024

phet Sep 25, 2024

phet Sep 25, 2024 •

edited

Loading

phet Sep 25, 2024

		PreparedStatement createStatement = connection.prepareStatement(
		String.format(CREATE_TABLE_STATEMENT, tableName))) {

[GOBBLIN-2137]merged dagNodeStateStore and failedDagNodeStateStore tables #4032

Are you sure you want to change the base?

[GOBBLIN-2137]merged dagNodeStateStore and failedDagNodeStateStore tables #4032

Conversation

pratapaditya04 commented Aug 16, 2024 • edited Loading

JIRA

Description

Tests

Commits

codecov-commenter commented Aug 19, 2024 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

arjun4084346 commented Aug 20, 2024

pratapaditya04 commented Aug 21, 2024

phet left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

phet Sep 25, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pratapaditya04 commented Aug 16, 2024 •

edited

Loading

codecov-commenter commented Aug 19, 2024 •

edited

Loading

phet Sep 25, 2024 •

edited

Loading