[jira] [Work logged] (GOBBLIN-790) Mysql Implementation of DagStateStore
[
https://issues.apache.org/jira/browse/GOBBLIN-790?focusedWorklogId=256610=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-256610
]
ASF GitHub Bot logged work on GOBBLIN-790:
--
Author: ASF GitHub Bot
Created on: 10/Jun/19 04:29
Start Date: 10/Jun/19 04:29
Worklog Time Spent: 10m
Work Description: sv2000 commented on pull request #2656:
[GOBBLIN-790]DagStateStore MySQL
URL: https://github.com/apache/incubator-gobblin/pull/2656#discussion_r291876727
##
File path:
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStore.java
##
@@ -0,0 +1,147 @@
+/*
+ * 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.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.metastore.MysqlStateStore;
+import org.apache.gobblin.metastore.MysqlStateStoreFactory;
+import org.apache.gobblin.metastore.StateStore;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.spec.GsonSerDe;
+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 com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import static
org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static
org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateFlowIdInString;
+import static
org.apache.gobblin.service.modules.orchestration.DagManagerUtils.getFlowExecId;
+
+
+/**
+ * A implementation of {@link DagStateStore} using MySQL as a backup, leverage
{@link MysqlStateStore}.
+ * It implements interfaces of {@link DagStateStore} but delegating
responsibilities to methods provided
+ * in {@link MysqlStateStore}.
+ * It also implements conversion between {@link Dag} to
{@link State}.
+ *
+ * The schema of this will simply be:
+ * | storeName | tableName | State |
+ * where storeName represents FlowId, a combination of FlowGroup and FlowName,
and tableName represents FlowExecutionId.
+ * State is a pocket for serialized {@link Dag} object.
+ *
+ * TODO: In the DagManagerTest: change the hardcoded type of DagStateStore.
+ *
+ */
+public class MysqlDagStateStore implements DagStateStore {
+
+ public static final String CONFIG_PREFIX = GOBBLIN_SERVICE_PREFIX +
"mysqlDagStateStore";
+ public static final String DAG_KEY_IN_STATE = "dag";
+
+ /**
+ * The schema of {@link MysqlStateStore} is fixed but the columns are
semantically projected into Dag's context:
+ * - The 'storeName' is FlowId.
+ * - The 'tableName' is FlowExecutionId.
+ */
+ private MysqlStateStore mysqlStateStore;
+ private final GsonSerDe> serDe;
+ private JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+ public MysqlDagStateStore(Config config, Map
topologySpecMap) {
Review comment:
Any chance we can just have a DagStateStoreImpl class that instantiates the
appropriate StateStore instance based on the config? Most of what you have
implemented is generic and can be used for any StateStore implementation. In
other words, we can replace FSDagStateStore and MysqlDagStateStore classes with
a single DagStateStoreImpl class.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id:
[jira] [Work logged] (GOBBLIN-790) Mysql Implementation of DagStateStore
[
https://issues.apache.org/jira/browse/GOBBLIN-790?focusedWorklogId=256607=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-256607
]
ASF GitHub Bot logged work on GOBBLIN-790:
--
Author: ASF GitHub Bot
Created on: 10/Jun/19 04:29
Start Date: 10/Jun/19 04:29
Worklog Time Spent: 10m
Work Description: sv2000 commented on pull request #2656:
[GOBBLIN-790]DagStateStore MySQL
URL: https://github.com/apache/incubator-gobblin/pull/2656#discussion_r291874816
##
File path:
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java
##
@@ -582,11 +581,18 @@ private void cleanUp() {
}
}
-private void cleanUpDag(String dagId) {
- Dag dag = this.dags.get(dagId);
- this.dagToJobs.remove(dagId);
- this.dagStateStore.cleanUp(dag);
+/**
+ * Cleaning of all relevant states need to be
+ * @param dagId
+ */
+private synchronized void cleanUpDag(String dagId) {
+ try {
+ this.dagStateStore.cleanUp(dags.get(dagId));
+ } catch (IOException ioe) {
+ log.error(String.format("Failed to clean %s from backStore due to:",
dagId), ioe);
+ }
this.dags.remove(dagId);
+ this.dagToJobs.remove(dagId);
Review comment:
Any reason why cleanUp(dag) and remove(dagId) are re-ordered? If the
ordering is important, make it clear in the comments.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 256607)
Time Spent: 0.5h (was: 20m)
> Mysql Implementation of DagStateStore
> -
>
> Key: GOBBLIN-790
> URL: https://issues.apache.org/jira/browse/GOBBLIN-790
> Project: Apache Gobblin
> Issue Type: Improvement
>Reporter: Lei Sun
>Priority: Major
> Time Spent: 0.5h
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (GOBBLIN-790) Mysql Implementation of DagStateStore
[
https://issues.apache.org/jira/browse/GOBBLIN-790?focusedWorklogId=256609=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-256609
]
ASF GitHub Bot logged work on GOBBLIN-790:
--
Author: ASF GitHub Bot
Created on: 10/Jun/19 04:29
Start Date: 10/Jun/19 04:29
Worklog Time Spent: 10m
Work Description: sv2000 commented on pull request #2656:
[GOBBLIN-790]DagStateStore MySQL
URL: https://github.com/apache/incubator-gobblin/pull/2656#discussion_r291874016
##
File path:
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java
##
@@ -582,11 +581,18 @@ private void cleanUp() {
}
}
-private void cleanUpDag(String dagId) {
- Dag dag = this.dags.get(dagId);
- this.dagToJobs.remove(dagId);
- this.dagStateStore.cleanUp(dag);
+/**
+ * Cleaning of all relevant states need to be
+ * @param dagId
+ */
+private synchronized void cleanUpDag(String dagId) {
+ try {
+ this.dagStateStore.cleanUp(dags.get(dagId));
+ } catch (IOException ioe) {
+ log.error(String.format("Failed to clean %s from backStore due to:",
dagId), ioe);
Review comment:
log.error("Failed to clean {} from backStore due to: ", dagId, ioe);?
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 256609)
Time Spent: 50m (was: 40m)
> Mysql Implementation of DagStateStore
> -
>
> Key: GOBBLIN-790
> URL: https://issues.apache.org/jira/browse/GOBBLIN-790
> Project: Apache Gobblin
> Issue Type: Improvement
>Reporter: Lei Sun
>Priority: Major
> Time Spent: 50m
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (GOBBLIN-790) Mysql Implementation of DagStateStore
[ https://issues.apache.org/jira/browse/GOBBLIN-790?focusedWorklogId=256608=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-256608 ] ASF GitHub Bot logged work on GOBBLIN-790: -- Author: ASF GitHub Bot Created on: 10/Jun/19 04:29 Start Date: 10/Jun/19 04:29 Worklog Time Spent: 10m Work Description: sv2000 commented on pull request #2656: [GOBBLIN-790]DagStateStore MySQL URL: https://github.com/apache/incubator-gobblin/pull/2656#discussion_r291873791 ## File path: gobblin-metastore/src/main/java/org/apache/gobblin/metastore/MysqlStateStore.java ## @@ -123,6 +125,7 @@ private final String UPSERT_JOB_STATE_SQL; private final String SELECT_JOB_STATE_SQL; + private final String SELECT_ALL_JOB_STATE_SQL; Review comment: rename variable to SELECT_ALL_JOBS_STATE_SQL? 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 256608) Time Spent: 40m (was: 0.5h) > Mysql Implementation of DagStateStore > - > > Key: GOBBLIN-790 > URL: https://issues.apache.org/jira/browse/GOBBLIN-790 > Project: Apache Gobblin > Issue Type: Improvement >Reporter: Lei Sun >Priority: Major > Time Spent: 40m > Remaining Estimate: 0h > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[GitHub] [incubator-gobblin] sv2000 commented on a change in pull request #2656: [GOBBLIN-790]DagStateStore MySQL
sv2000 commented on a change in pull request #2656: [GOBBLIN-790]DagStateStore MySQL URL: https://github.com/apache/incubator-gobblin/pull/2656#discussion_r291873791 ## File path: gobblin-metastore/src/main/java/org/apache/gobblin/metastore/MysqlStateStore.java ## @@ -123,6 +125,7 @@ private final String UPSERT_JOB_STATE_SQL; private final String SELECT_JOB_STATE_SQL; + private final String SELECT_ALL_JOB_STATE_SQL; Review comment: rename variable to SELECT_ALL_JOBS_STATE_SQL? 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [incubator-gobblin] sv2000 commented on a change in pull request #2656: [GOBBLIN-790]DagStateStore MySQL
sv2000 commented on a change in pull request #2656: [GOBBLIN-790]DagStateStore
MySQL
URL: https://github.com/apache/incubator-gobblin/pull/2656#discussion_r291876727
##
File path:
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlDagStateStore.java
##
@@ -0,0 +1,147 @@
+/*
+ * 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.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.metastore.MysqlStateStore;
+import org.apache.gobblin.metastore.MysqlStateStoreFactory;
+import org.apache.gobblin.metastore.StateStore;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.spec.GsonSerDe;
+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 com.google.gson.JsonDeserializer;
+import com.google.gson.JsonSerializer;
+import com.google.gson.reflect.TypeToken;
+import com.typesafe.config.Config;
+
+import static
org.apache.gobblin.service.ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX;
+import static
org.apache.gobblin.service.modules.orchestration.DagManagerUtils.generateFlowIdInString;
+import static
org.apache.gobblin.service.modules.orchestration.DagManagerUtils.getFlowExecId;
+
+
+/**
+ * A implementation of {@link DagStateStore} using MySQL as a backup, leverage
{@link MysqlStateStore}.
+ * It implements interfaces of {@link DagStateStore} but delegating
responsibilities to methods provided
+ * in {@link MysqlStateStore}.
+ * It also implements conversion between {@link Dag} to
{@link State}.
+ *
+ * The schema of this will simply be:
+ * | storeName | tableName | State |
+ * where storeName represents FlowId, a combination of FlowGroup and FlowName,
and tableName represents FlowExecutionId.
+ * State is a pocket for serialized {@link Dag} object.
+ *
+ * TODO: In the DagManagerTest: change the hardcoded type of DagStateStore.
+ *
+ */
+public class MysqlDagStateStore implements DagStateStore {
+
+ public static final String CONFIG_PREFIX = GOBBLIN_SERVICE_PREFIX +
"mysqlDagStateStore";
+ public static final String DAG_KEY_IN_STATE = "dag";
+
+ /**
+ * The schema of {@link MysqlStateStore} is fixed but the columns are
semantically projected into Dag's context:
+ * - The 'storeName' is FlowId.
+ * - The 'tableName' is FlowExecutionId.
+ */
+ private MysqlStateStore mysqlStateStore;
+ private final GsonSerDe> serDe;
+ private JobExecutionPlanDagFactory jobExecPlanDagFactory;
+
+ public MysqlDagStateStore(Config config, Map
topologySpecMap) {
Review comment:
Any chance we can just have a DagStateStoreImpl class that instantiates the
appropriate StateStore instance based on the config? Most of what you have
implemented is generic and can be used for any StateStore implementation. In
other words, we can replace FSDagStateStore and MysqlDagStateStore classes with
a single DagStateStoreImpl class.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-gobblin] sv2000 commented on a change in pull request #2656: [GOBBLIN-790]DagStateStore MySQL
sv2000 commented on a change in pull request #2656: [GOBBLIN-790]DagStateStore
MySQL
URL: https://github.com/apache/incubator-gobblin/pull/2656#discussion_r291874016
##
File path:
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java
##
@@ -582,11 +581,18 @@ private void cleanUp() {
}
}
-private void cleanUpDag(String dagId) {
- Dag dag = this.dags.get(dagId);
- this.dagToJobs.remove(dagId);
- this.dagStateStore.cleanUp(dag);
+/**
+ * Cleaning of all relevant states need to be
+ * @param dagId
+ */
+private synchronized void cleanUpDag(String dagId) {
+ try {
+ this.dagStateStore.cleanUp(dags.get(dagId));
+ } catch (IOException ioe) {
+ log.error(String.format("Failed to clean %s from backStore due to:",
dagId), ioe);
Review comment:
log.error("Failed to clean {} from backStore due to: ", dagId, ioe);?
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [incubator-gobblin] htran1 commented on a change in pull request #2665: [GOBBLIN-798] Clean up workflows from Helix when the Gobblin applicat…
htran1 commented on a change in pull request #2665: [GOBBLIN-798] Clean up workflows from Helix when the Gobblin applicat… URL: https://github.com/apache/incubator-gobblin/pull/2665#discussion_r291871752 ## File path: gobblin-cluster/src/main/java/org/apache/gobblin/cluster/GobblinClusterConfigurationKeys.java ## @@ -161,4 +161,8 @@ public static final String CANCEL_RUNNING_JOB_ON_DELETE = GOBBLIN_CLUSTER_PREFIX + "job.cancelRunningJobOnDelete"; public static final String DEFAULT_CANCEL_RUNNING_JOB_ON_DELETE = "false"; + + // for cleaning up jobs on cluster manager startup + public static final String CLEAN_UP_JOBS_ON_MANAGER_START = GOBBLIN_CLUSTER_PREFIX + "cleanUpJobsOnManagerStart"; + public static final boolean DEFAULT_CLEAN_UP_JOBS_ON_MANAGER_START = false; Review comment: Removed this option since the existing behavior of not cleaning up on startup lead to the issue being fixed in this PR, so it is reasonable to not have an option to preserve the buggy behavior. 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
