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/31 21:48:09 UTC
[PR] redirect kill requests to dag proc engine [gobblin]
arjun4084346 opened a new pull request, #4012:
URL: https://github.com/apache/gobblin/pull/4012
do not retry non-retryable exceptions
.
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] [GOBBLIN-2121] redirect kill requests to dag proc engine [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #4012:
URL: https://github.com/apache/gobblin/pull/4012#discussion_r1700868425
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -319,14 +322,30 @@ public void submitFlowToDagManager(FlowSpec flowSpec, Dag<JobExecutionPlan> jobE
}
public void remove(Spec spec, Properties headers) throws IOException {
+ URI uri = spec.getUri();
// TODO: Evolve logic to cache and reuse previously compiled JobSpecs
// .. this will work for Identity compiler but not always for multi-hop.
// Note: Current logic assumes compilation is consistent between all executions
if (spec instanceof FlowSpec) {
- //Send the dag to the DagManager to stop it.
- //Also send it to the SpecProducer to do any cleanup tasks on SpecExecutor.
- _log.info("Forwarding cancel request for flow URI {} to DagManager.", spec.getUri());
- this.dagManager.stopDag(spec.getUri());
+ String flowGroup = FlowSpec.Utils.getFlowGroup(uri);
+ String flowName = FlowSpec.Utils.getFlowName(uri);
+ if (this.flowLaunchHandler.isPresent()) {
+ List<Long> flowExecutionIds = this.jobStatusRetriever.getLatestExecutionIdsForFlow(flowName, flowGroup, 10);
+ _log.info("Found {} flows to cancel.", flowExecutionIds.size());
+
+ for (long flowExecutionId : flowExecutionIds) {
+ DagActionStore.DagAction killDagAction = DagActionStore.DagAction.forFlow(flowGroup, flowName, flowExecutionId,
+ DagActionStore.DagActionType.KILL);
Review Comment:
yea, it is about killing all flows if any. assuming 10 flows to cancel...just copied from DagManager code
--
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-2121] redirect kill requests to dag proc engine [gobblin]
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #4012:
URL: https://github.com/apache/gobblin/pull/4012#discussion_r1700753986
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagActionStore.java:
##########
@@ -33,7 +33,7 @@
* See javadoc for {@link DagAction}
*/
public interface DagActionStore {
- public static final String NO_JOB_NAME_DEFAULT = "";
+ String NO_JOB_NAME_DEFAULT = "";
Review Comment:
why not static final?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/FlowLaunchHandler.java:
##########
@@ -120,6 +120,24 @@ && persistLaunchDagAction((LeaseAttemptStatus.LeaseObtainedStatus) leaseAttempt)
}
}
+ /**
+ * This method is used in the multi-active scheduler case for one or more hosts to respond to a kill dag action
+ * event triggered by the Orchestrator by attempting a lease for the kill event and processing the result depending on
+ * the status of the attempt.
+ */
+ public void handleFlowKillTriggerEvent(Properties jobProps, DagActionStore.LeaseParams leaseParams) throws IOException {
Review Comment:
do we need this for resume as well? also this has duplicate code with launch so we can probably extract them out and have a function with the common logic that both `handleFlowLaunchTriggerEvent` & `handleFlowKillTriggerEvent` utilize
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngine.java:
##########
@@ -140,6 +153,11 @@ public void run() {
dagTask.conclude();
} catch (Exception e) {
log.error("DagProcEngineThread encountered exception while processing dag " + dagProc.getDagId(), e);
+ if (KafkaJobStatusMonitor.isThrowableInstanceOf(e, this.nonRetryableExceptions)) {
+ // conclude the lease so that it is not retried, if the dag proc fails with non-transient exception
+ dagTask.conclude();
+ dagManagementStateStore.getDagManagerMetrics().dagProcessingNonRetryableExceptionMeter.mark();
+ }
dagManagementStateStore.getDagManagerMetrics().dagProcessingExceptionMeter.mark();
Review Comment:
this should be in else block otherwise we are double counting
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagerMetrics.java:
##########
@@ -103,6 +104,8 @@ public void activate() {
ServiceMetricNames.JOBS_SENT_TO_SPEC_EXECUTOR));
dagProcessingExceptionMeter = metricContext.contextAwareMeter(MetricRegistry.name(ServiceMetricNames.GOBBLIN_SERVICE_PREFIX,
ServiceMetricNames.DAG_PROCESSING_EXCEPTION_METER));
+ dagProcessingNonRetryableExceptionMeter = metricContext.contextAwareMeter(MetricRegistry.name(ServiceMetricNames.GOBBLIN_SERVICE_PREFIX,
+ ServiceMetricNames.DAG_PROCESSING_NON_RETRYABLE_EXCEPTION_METER));
Review Comment:
should we move both to DagProcessingEngineMetrics or do want to do in a later PR where the two are merged?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -319,14 +322,30 @@ public void submitFlowToDagManager(FlowSpec flowSpec, Dag<JobExecutionPlan> jobE
}
public void remove(Spec spec, Properties headers) throws IOException {
+ URI uri = spec.getUri();
// TODO: Evolve logic to cache and reuse previously compiled JobSpecs
// .. this will work for Identity compiler but not always for multi-hop.
// Note: Current logic assumes compilation is consistent between all executions
if (spec instanceof FlowSpec) {
- //Send the dag to the DagManager to stop it.
- //Also send it to the SpecProducer to do any cleanup tasks on SpecExecutor.
- _log.info("Forwarding cancel request for flow URI {} to DagManager.", spec.getUri());
- this.dagManager.stopDag(spec.getUri());
+ String flowGroup = FlowSpec.Utils.getFlowGroup(uri);
+ String flowName = FlowSpec.Utils.getFlowName(uri);
+ if (this.flowLaunchHandler.isPresent()) {
+ List<Long> flowExecutionIds = this.jobStatusRetriever.getLatestExecutionIdsForFlow(flowName, flowGroup, 10);
+ _log.info("Found {} flows to cancel.", flowExecutionIds.size());
+
+ for (long flowExecutionId : flowExecutionIds) {
+ DagActionStore.DagAction killDagAction = DagActionStore.DagAction.forFlow(flowGroup, flowName, flowExecutionId,
+ DagActionStore.DagActionType.KILL);
Review Comment:
is this only for case that we kill previous executions of a flow before a launch? not user invoked kill? why are we calling this in the orchestrator asfaik it should only deal with launches
##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngineTest.java:
##########
@@ -76,26 +75,21 @@ public void setUp() throws Exception {
.addPrimitive(MysqlUserQuotaManager.qualify(ConfigurationKeys.STATE_STORE_DB_URL_KEY), testMetastoreDatabase.getJdbcUrl())
.addPrimitive(MysqlUserQuotaManager.qualify(ConfigurationKeys.STATE_STORE_DB_USER_KEY), TEST_USER)
.addPrimitive(MysqlUserQuotaManager.qualify(ConfigurationKeys.STATE_STORE_DB_PASSWORD_KEY), TEST_PASSWORD)
- .addPrimitive(MysqlUserQuotaManager.qualify(ConfigurationKeys.STATE_STORE_DB_TABLE_KEY), TEST_TABLE);
+ .addPrimitive(MysqlUserQuotaManager.qualify(ConfigurationKeys.STATE_STORE_DB_TABLE_KEY), TEST_TABLE)
+ .addPrimitive(ServiceConfigKeys.DAG_PROC_ENGINE_NON_RETRYABLE_EXCEPTIONS_KEY, NonTransientException.class.getName());
Review Comment:
how are NonTransientExceptions classified? Can we test specifically with the ServiceAcl error
--
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-2121] redirect kill requests to dag proc engine [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #4012:
URL: https://github.com/apache/gobblin/pull/4012#discussion_r1700867305
##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngineTest.java:
##########
@@ -76,26 +75,21 @@ public void setUp() throws Exception {
.addPrimitive(MysqlUserQuotaManager.qualify(ConfigurationKeys.STATE_STORE_DB_URL_KEY), testMetastoreDatabase.getJdbcUrl())
.addPrimitive(MysqlUserQuotaManager.qualify(ConfigurationKeys.STATE_STORE_DB_USER_KEY), TEST_USER)
.addPrimitive(MysqlUserQuotaManager.qualify(ConfigurationKeys.STATE_STORE_DB_PASSWORD_KEY), TEST_PASSWORD)
- .addPrimitive(MysqlUserQuotaManager.qualify(ConfigurationKeys.STATE_STORE_DB_TABLE_KEY), TEST_TABLE);
+ .addPrimitive(MysqlUserQuotaManager.qualify(ConfigurationKeys.STATE_STORE_DB_TABLE_KEY), TEST_TABLE)
+ .addPrimitive(ServiceConfigKeys.DAG_PROC_ENGINE_NON_RETRYABLE_EXCEPTIONS_KEY, NonTransientException.class.getName());
Review Comment:
NonTransientException.class is just a exception class in oss gobblin. ServiceAcl is in ligobblin.
--
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-2121] redirect kill requests to dag proc engine [gobblin]
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #4012:
URL: https://github.com/apache/gobblin/pull/4012#discussion_r1701585181
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/FlowLaunchHandler.java:
##########
Review Comment:
let's keep the PRs as small and as cohesive as possible. so next time, I'd strongly encourage you to put this and related mods into a PR separate from something like the transient exceptions work change. they seem orthogonal
--
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-2121] redirect kill requests to dag proc engine [gobblin]
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #4012:
URL: https://github.com/apache/gobblin/pull/4012#discussion_r1701588762
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/DeadlineEnforcementDagProc.java:
##########
@@ -68,7 +68,7 @@ protected boolean isDagStillPresent(Optional<Dag<JobExecutionPlan>> dag, DagMana
if (!dagManagementStateStore.existsJobDagAction(dagAction.getFlowGroup(), dagAction.getFlowName(),
dagAction.getFlowExecutionId(), dagAction.getJobName(), dagAction.getDagActionType())) {
- log.warn("Dag action {} is cleaned up from DMSS. No further action is required.", dagAction);
+ log.info("Dag action {} is cleaned up from DMSS. No further action is required.", dagAction);
return false;
Review Comment:
don't we already hold a lease by the time `act` is called? if so, it would be exceedingly rare (and really no different than any other `DagProc::act` derived class impl). given the rarity, I'd suggest to remove.
but if we really believe we need this, let's observe the commonality by placing this impl within the `DagProc` base 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.
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-2121] redirect kill requests to dag proc engine [gobblin]
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #4012:
URL: https://github.com/apache/gobblin/pull/4012#discussion_r1702273840
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngine.java:
##########
@@ -140,6 +153,11 @@ public void run() {
dagTask.conclude();
} catch (Exception e) {
log.error("DagProcEngineThread encountered exception while processing dag " + dagProc.getDagId(), e);
+ if (KafkaJobStatusMonitor.isThrowableInstanceOf(e, this.nonRetryableExceptions)) {
+ // conclude the lease so that it is not retried, if the dag proc fails with non-transient exception
+ dagTask.conclude();
+ dagManagementStateStore.getDagManagerMetrics().dagProcessingNonRetryableExceptionMeter.mark();
Review Comment:
good point. really I'm just seeking better encapsulation within `DagProc`, so the DPE remains blissfully unaware of what they're doing and hence what exceptions they might throw.
most ideal might be each `DagProc` class to know which of its exceptions are transient. e.g. a base class method `boolean isTransientException(Throwable t)`, with a common base class version and any derived class extending beyond that (while delegating to the base class impl as a fallback).
if for expedience, you want a small step in the right direction that would be for the factory to be initialized w/ a predicate as I suggested - given it's the one who decides which specific `DagProc` derived class impls to use.
but after consider further, I really think `DagProc::isTransientException` is preferable.
--
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-2121] redirect kill requests to dag proc engine [gobblin]
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #4012:
URL: https://github.com/apache/gobblin/pull/4012#discussion_r1700916336
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -319,14 +322,30 @@ public void submitFlowToDagManager(FlowSpec flowSpec, Dag<JobExecutionPlan> jobE
}
public void remove(Spec spec, Properties headers) throws IOException {
+ URI uri = spec.getUri();
// TODO: Evolve logic to cache and reuse previously compiled JobSpecs
// .. this will work for Identity compiler but not always for multi-hop.
// Note: Current logic assumes compilation is consistent between all executions
if (spec instanceof FlowSpec) {
- //Send the dag to the DagManager to stop it.
- //Also send it to the SpecProducer to do any cleanup tasks on SpecExecutor.
- _log.info("Forwarding cancel request for flow URI {} to DagManager.", spec.getUri());
- this.dagManager.stopDag(spec.getUri());
+ String flowGroup = FlowSpec.Utils.getFlowGroup(uri);
+ String flowName = FlowSpec.Utils.getFlowName(uri);
+ if (this.flowLaunchHandler.isPresent()) {
+ List<Long> flowExecutionIds = this.jobStatusRetriever.getLatestExecutionIdsForFlow(flowName, flowGroup, 10);
+ _log.info("Found {} flows to cancel.", flowExecutionIds.size());
+
+ for (long flowExecutionId : flowExecutionIds) {
+ DagActionStore.DagAction killDagAction = DagActionStore.DagAction.forFlow(flowGroup, flowName, flowExecutionId,
+ DagActionStore.DagActionType.KILL);
Review Comment:
I see this is only for delete flowSpec then it makes sense
--
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-2121] redirect kill requests to dag proc engine [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #4012:
URL: https://github.com/apache/gobblin/pull/4012#discussion_r1702197031
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngine.java:
##########
@@ -140,6 +153,11 @@ public void run() {
dagTask.conclude();
} catch (Exception e) {
log.error("DagProcEngineThread encountered exception while processing dag " + dagProc.getDagId(), e);
+ if (KafkaJobStatusMonitor.isThrowableInstanceOf(e, this.nonRetryableExceptions)) {
+ // conclude the lease so that it is not retried, if the dag proc fails with non-transient exception
+ dagTask.conclude();
+ dagManagementStateStore.getDagManagerMetrics().dagProcessingNonRetryableExceptionMeter.mark();
Review Comment:
why to pass another parameter to DagProcFactory or individual DagProcs?
We can just catch it in DagProc::process
--
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-2121] redirect kill requests to dag proc engine [gobblin]
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #4012:
URL: https://github.com/apache/gobblin/pull/4012#discussion_r1701592323
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/KafkaJobStatusMonitor.java:
##########
@@ -426,7 +426,15 @@ public static long getExecutionIdFromTableName(String tableName) {
protected abstract org.apache.gobblin.configuration.State parseJobStatus(GobblinTrackingEvent event);
- public static boolean isThrowableInstanceOf(Throwable exception, List<Class<? extends Exception>> typesList) {
- return typesList.stream().anyMatch(e -> e.isInstance(exception));
+ public static boolean isThrowableInstanceOf(Throwable exception, List<Class<? extends Exception>> exceptionsList) {
Review Comment:
this changes the semantics. perhaps that's OK, but a more precise name along w/ some javadoc might help. e.g. `isThrowableOrChildCauseInstanceOf` or `isThrowableInstanceOfRecursive`
also, seems better suited to living within a utils class, rather than here in KJSM which was probably merely the first place we realized we needed such functionality.
--
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-2121] redirect kill requests to dag proc engine [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 closed pull request #4012: [GOBBLIN-2121] redirect kill requests to dag proc engine
URL: https://github.com/apache/gobblin/pull/4012
--
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-2121] redirect kill requests to dag proc engine [gobblin]
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #4012:
URL: https://github.com/apache/gobblin/pull/4012#discussion_r1701579382
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngine.java:
##########
@@ -140,6 +153,11 @@ public void run() {
dagTask.conclude();
} catch (Exception e) {
log.error("DagProcEngineThread encountered exception while processing dag " + dagProc.getDagId(), e);
+ if (KafkaJobStatusMonitor.isThrowableInstanceOf(e, this.nonRetryableExceptions)) {
+ // conclude the lease so that it is not retried, if the dag proc fails with non-transient exception
Review Comment:
suggest to log (although better would be best contained within the predicate impl (i.e. `Throwable -> boolean`)
--
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-2121] redirect kill requests to dag proc engine [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #4012:
URL: https://github.com/apache/gobblin/pull/4012#discussion_r1700785027
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagerMetrics.java:
##########
@@ -103,6 +104,8 @@ public void activate() {
ServiceMetricNames.JOBS_SENT_TO_SPEC_EXECUTOR));
dagProcessingExceptionMeter = metricContext.contextAwareMeter(MetricRegistry.name(ServiceMetricNames.GOBBLIN_SERVICE_PREFIX,
ServiceMetricNames.DAG_PROCESSING_EXCEPTION_METER));
+ dagProcessingNonRetryableExceptionMeter = metricContext.contextAwareMeter(MetricRegistry.name(ServiceMetricNames.GOBBLIN_SERVICE_PREFIX,
+ ServiceMetricNames.DAG_PROCESSING_NON_RETRYABLE_EXCEPTION_METER));
Review Comment:
yes, later pr. that change will touch 30+ files
--
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-2121] redirect kill requests to dag proc engine [gobblin]
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #4012:
URL: https://github.com/apache/gobblin/pull/4012#discussion_r1702273840
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngine.java:
##########
@@ -140,6 +153,11 @@ public void run() {
dagTask.conclude();
} catch (Exception e) {
log.error("DagProcEngineThread encountered exception while processing dag " + dagProc.getDagId(), e);
+ if (KafkaJobStatusMonitor.isThrowableInstanceOf(e, this.nonRetryableExceptions)) {
+ // conclude the lease so that it is not retried, if the dag proc fails with non-transient exception
+ dagTask.conclude();
+ dagManagementStateStore.getDagManagerMetrics().dagProcessingNonRetryableExceptionMeter.mark();
Review Comment:
good point. really I'm just seeking better encapsulation within `DagProc`, so the DPE remains blissfully unaware of what they're doing and hence what exceptions they might throw.
most ideal might be each `DagProc` class to know which of its exceptions are transient. e.g. a base class method `boolean isTransientException(Throwable t)`, with a common base class version and any derived class extending beyond that (while delegating to the base class impl as a fallback).
if for expedience, you want a small step in the right direction that would be for the factory to be initialized w/ a predicate as I suggested - given it's the one who decides which specific `DagProc` derived class impls to use.
but after considering further, I really think `DagProc::isTransientException` is preferable.
--
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-2121] redirect kill requests to dag proc engine [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #4012:
URL: https://github.com/apache/gobblin/pull/4012#discussion_r1700784146
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagActionStore.java:
##########
@@ -33,7 +33,7 @@
* See javadoc for {@link DagAction}
*/
public interface DagActionStore {
- public static final String NO_JOB_NAME_DEFAULT = "";
+ String NO_JOB_NAME_DEFAULT = "";
Review Comment:
that's not needed for interface. they are always public static
--
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-2121] redirect kill requests to dag proc engine [gobblin]
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #4012:
URL: https://github.com/apache/gobblin/pull/4012#discussion_r1701582523
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -319,14 +322,30 @@ public void submitFlowToDagManager(FlowSpec flowSpec, Dag<JobExecutionPlan> jobE
}
public void remove(Spec spec, Properties headers) throws IOException {
+ URI uri = spec.getUri();
// TODO: Evolve logic to cache and reuse previously compiled JobSpecs
// .. this will work for Identity compiler but not always for multi-hop.
// Note: Current logic assumes compilation is consistent between all executions
if (spec instanceof FlowSpec) {
- //Send the dag to the DagManager to stop it.
- //Also send it to the SpecProducer to do any cleanup tasks on SpecExecutor.
- _log.info("Forwarding cancel request for flow URI {} to DagManager.", spec.getUri());
- this.dagManager.stopDag(spec.getUri());
+ String flowGroup = FlowSpec.Utils.getFlowGroup(uri);
+ String flowName = FlowSpec.Utils.getFlowName(uri);
+ if (this.flowLaunchHandler.isPresent()) {
+ List<Long> flowExecutionIds = this.jobStatusRetriever.getLatestExecutionIdsForFlow(flowName, flowGroup, 10);
+ _log.info("Found {} flows to cancel.", flowExecutionIds.size());
+
+ for (long flowExecutionId : flowExecutionIds) {
+ DagActionStore.DagAction killDagAction = DagActionStore.DagAction.forFlow(flowGroup, flowName, flowExecutionId,
+ DagActionStore.DagActionType.KILL);
+ DagActionStore.LeaseParams leaseObject = new DagActionStore.LeaseParams(killDagAction, false,
Review Comment:
nit: `leaseParams` rather than `leaseObject`
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/FlowLaunchHandler.java:
##########
Review Comment:
let's keep the PRs as small and as cohesive as possible. so next time, I'd strongly encourage you to put this and related mods into a PR separate from something like the non-transient exceptions work
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngine.java:
##########
@@ -140,6 +153,11 @@ public void run() {
dagTask.conclude();
} catch (Exception e) {
log.error("DagProcEngineThread encountered exception while processing dag " + dagProc.getDagId(), e);
+ if (KafkaJobStatusMonitor.isThrowableInstanceOf(e, this.nonRetryableExceptions)) {
+ // conclude the lease so that it is not retried, if the dag proc fails with non-transient exception
Review Comment:
suggest to log (although better would be within the predicate impl (i.e. `Exception -> boolean`)
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngine.java:
##########
@@ -122,6 +134,7 @@ static class DagProcEngineThread implements Runnable {
private final DagProcFactory dagProcFactory;
private final DagManagementStateStore dagManagementStateStore;
private final DagProcessingEngineMetrics dagProcEngineMetrics;
+ private final List<Class<? extends Exception>> nonRetryableExceptions;
Review Comment:
let's use a predicate as the interface - i.e. a function of `Exception -> boolean`, but you could certainly implement one instance of that Strategy interface in terms of a `List<Class<? extends Exception>>`
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/KafkaJobStatusMonitor.java:
##########
@@ -426,7 +426,15 @@ public static long getExecutionIdFromTableName(String tableName) {
protected abstract org.apache.gobblin.configuration.State parseJobStatus(GobblinTrackingEvent event);
- public static boolean isThrowableInstanceOf(Throwable exception, List<Class<? extends Exception>> typesList) {
- return typesList.stream().anyMatch(e -> e.isInstance(exception));
+ public static boolean isThrowableInstanceOf(Throwable exception, List<Class<? extends Exception>> exceptionsList) {
Review Comment:
this changes the semantics. perhaps that's OK, but a more precise name along w/ some javadoc might help. e.g. `isThrowableOrCaseInstanceOf` or `isThrowableInstanceOfRecursive`
also, seems better suited to living within a utils class, rather than here in KJSM which was probably just the first place we realized we needed such fucntionality.
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngine.java:
##########
@@ -140,6 +153,11 @@ public void run() {
dagTask.conclude();
} catch (Exception e) {
log.error("DagProcEngineThread encountered exception while processing dag " + dagProc.getDagId(), e);
+ if (KafkaJobStatusMonitor.isThrowableInstanceOf(e, this.nonRetryableExceptions)) {
+ // conclude the lease so that it is not retried, if the dag proc fails with non-transient exception
+ dagTask.conclude();
+ dagManagementStateStore.getDagManagerMetrics().dagProcessingNonRetryableExceptionMeter.mark();
Review Comment:
couldn't this live entirely encapsulated within `DagProc::process`? that would give a normal return, which means the DPE would next call `DagTask::conclude`
for that approach, initialize the `DagProcFactory` w/ the exception predicate
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/DeadlineEnforcementDagProc.java:
##########
@@ -68,7 +68,7 @@ protected boolean isDagStillPresent(Optional<Dag<JobExecutionPlan>> dag, DagMana
if (!dagManagementStateStore.existsJobDagAction(dagAction.getFlowGroup(), dagAction.getFlowName(),
dagAction.getFlowExecutionId(), dagAction.getJobName(), dagAction.getDagActionType())) {
- log.warn("Dag action {} is cleaned up from DMSS. No further action is required.", dagAction);
+ log.info("Dag action {} is cleaned up from DMSS. No further action is required.", dagAction);
return false;
Review Comment:
don't we already hold a lease by the time `act` is called? if so, it would be exceedingly rare (and really no different than any other `DagProc::act` derived class impl). hence, I'd suggest to remove. but if we really believe we need this, let's put the impl into the `DagProc` base 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.
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-2121] redirect kill requests to dag proc engine [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #4012:
URL: https://github.com/apache/gobblin/pull/4012#discussion_r1700785719
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngine.java:
##########
@@ -140,6 +153,11 @@ public void run() {
dagTask.conclude();
} catch (Exception e) {
log.error("DagProcEngineThread encountered exception while processing dag " + dagProc.getDagId(), e);
+ if (KafkaJobStatusMonitor.isThrowableInstanceOf(e, this.nonRetryableExceptions)) {
+ // conclude the lease so that it is not retried, if the dag proc fails with non-transient exception
+ dagTask.conclude();
+ dagManagementStateStore.getDagManagerMetrics().dagProcessingNonRetryableExceptionMeter.mark();
+ }
dagManagementStateStore.getDagManagerMetrics().dagProcessingExceptionMeter.mark();
Review Comment:
yes, we want to double count, because every non-retryable exception is still an exception
--
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-2121] redirect kill requests to dag proc engine [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #4012:
URL: https://github.com/apache/gobblin/pull/4012#discussion_r1700888520
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/FlowLaunchHandler.java:
##########
@@ -120,6 +120,24 @@ && persistLaunchDagAction((LeaseAttemptStatus.LeaseObtainedStatus) leaseAttempt)
}
}
+ /**
+ * This method is used in the multi-active scheduler case for one or more hosts to respond to a kill dag action
+ * event triggered by the Orchestrator by attempting a lease for the kill event and processing the result depending on
+ * the status of the attempt.
+ */
+ public void handleFlowKillTriggerEvent(Properties jobProps, DagActionStore.LeaseParams leaseParams) throws IOException {
Review Comment:
delete and resume rest commands are actually already creating dag actions as expected, check FlowExecutionResourceLocalHandler.
this dag kill this PR is addressing is about the expectation that when a flowConfig is deleted, its dags should also be stopped.
--
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-2121] redirect kill requests to dag proc engine [gobblin]
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #4012:
URL: https://github.com/apache/gobblin/pull/4012#discussion_r1701578156
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngine.java:
##########
@@ -122,6 +134,7 @@ static class DagProcEngineThread implements Runnable {
private final DagProcFactory dagProcFactory;
private final DagManagementStateStore dagManagementStateStore;
private final DagProcessingEngineMetrics dagProcEngineMetrics;
+ private final List<Class<? extends Exception>> nonRetryableExceptions;
Review Comment:
let's use a predicate as the interface - i.e. a function of `Throwable -> boolean`, but you could certainly implement one instance of such a Strategy interface in terms of a `List<Class<? extends Exception>>`
--
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-2121] redirect kill requests to dag proc engine [gobblin]
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #4012:
URL: https://github.com/apache/gobblin/pull/4012#discussion_r1701598075
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngine.java:
##########
@@ -140,6 +153,11 @@ public void run() {
dagTask.conclude();
} catch (Exception e) {
log.error("DagProcEngineThread encountered exception while processing dag " + dagProc.getDagId(), e);
+ if (KafkaJobStatusMonitor.isThrowableInstanceOf(e, this.nonRetryableExceptions)) {
+ // conclude the lease so that it is not retried, if the dag proc fails with non-transient exception
+ dagTask.conclude();
+ dagManagementStateStore.getDagManagerMetrics().dagProcessingNonRetryableExceptionMeter.mark();
Review Comment:
couldn't this live entirely encapsulated within `DagProc::process`? let it make a normal return, so the DPE would continue on to `DagTask::conclude`
(for that approach, initialize the `DagProcFactory` w/ the exception predicate discussed above)
--
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