You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gobblin.apache.org by ar...@apache.org on 2024/03/19 16:44:37 UTC
(gobblin) branch master updated: Generalize LeaseArbiterDecorator for Execution & Scheduling (#3890)
This is an automated email from the ASF dual-hosted git repository.
arjun4084346 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/gobblin.git
The following commit(s) were added to refs/heads/master by this push:
new 8066dc48a Generalize LeaseArbiterDecorator for Execution & Scheduling (#3890)
8066dc48a is described below
commit 8066dc48a14d15fd3014c0d598de685478000922
Author: umustafi <um...@gmail.com>
AuthorDate: Tue Mar 19 09:44:29 2024 -0700
Generalize LeaseArbiterDecorator for Execution & Scheduling (#3890)
* Define general & dagProc lease arbitration handlers
* Framework for reminder and quartz scheduler
* Construct Reminder JobDetail & Trigger
* Initialize Quartz scheduler and lease arbiter handlers
* Use a common Quartz SchedulerFactory to create scheduler
* Cleanup unused code
* Add more TODO statements for lease arbiter
* Add javadocs & cleanup some functions
* Utilize decorator pattern, initialize SchedulerService correctly
* also renames flowAction to dagAction
* clarifies naming for decorator classes
* Add job_name to dagActionStore & leaseArbiter schemas
* Change FlowTrigger to be a handler instead of decorator
* address other review comments
* Add javadoc to ReminderJob class and methods
* Correct findBug errors
* Fix failing unit tests
* Initialize dagProcessingEngine & multi-active execution classes with Guice & inject properly
* Respond to review comments
---------
Co-authored-by: Urmi Mustafi <um...@linkedin.com>
---
.../apache/gobblin/service/ServiceConfigKeys.java | 1 +
.../apache/gobblin/metrics/ServiceMetricNames.java | 3 +-
.../apache/gobblin/runtime/api/DagActionStore.java | 40 +++-
.../runtime/api/MultiActiveLeaseArbiter.java | 68 ++++---
.../runtime/api/MysqlMultiActiveLeaseArbiter.java | 202 +++++++++++----------
.../dag_action_store/MysqlDagActionStore.java | 60 +++---
.../gobblin/runtime/util/InjectionNames.java | 1 +
.../apache/gobblin/scheduler/SchedulerService.java | 36 +++-
.../api/MysqlMultiActiveLeaseArbiterTest.java | 34 ++--
.../dag_action_store/MysqlDagActionStoreTest.java | 35 ++--
.../modules/core/GobblinServiceConfiguration.java | 4 +
.../modules/core/GobblinServiceGuiceModule.java | 45 ++++-
.../orchestration/DagActionReminderScheduler.java | 76 ++++++++
.../orchestration/DagManagementTaskStreamImpl.java | 34 ++--
.../service/modules/orchestration/DagManager.java | 19 +-
.../modules/orchestration/DagManagerUtils.java | 11 +-
.../modules/orchestration/DagProcessingEngine.java | 16 +-
...wTriggerHandler.java => FlowLaunchHandler.java} | 149 ++++++---------
.../orchestration/InstrumentedLeaseArbiter.java | 112 ++++++++++++
.../modules/orchestration/Orchestrator.java | 24 ++-
.../ReminderSettingDagProcLeaseArbiter.java | 199 ++++++++++++++++++++
.../modules/orchestration/proc/LaunchDagProc.java | 1 -
.../modules/orchestration/task/DagTask.java | 9 +-
.../modules/orchestration/task/LaunchDagTask.java | 6 -
...lowExecutionResourceHandlerWithWarmStandby.java | 6 +-
.../scheduler/GobblinServiceJobScheduler.java | 14 +-
.../monitoring/DagActionStoreChangeMonitor.java | 17 +-
.../DagManagementDagActionStoreChangeMonitor.java | 6 +-
.../DagManagementTaskStreamImplTest.java | 35 +++-
.../modules/orchestration/DagManagerFlowTest.java | 16 +-
.../orchestration/DagProcessingEngineTest.java | 9 +-
...HandlerTest.java => FlowLaunchHandlerTest.java} | 16 +-
.../modules/orchestration/OrchestratorTest.java | 4 +-
.../orchestration/proc/LaunchDagProcTest.java | 2 +-
.../scheduler/GobblinServiceJobSchedulerTest.java | 9 +-
35 files changed, 923 insertions(+), 396 deletions(-)
diff --git a/gobblin-api/src/main/java/org/apache/gobblin/service/ServiceConfigKeys.java b/gobblin-api/src/main/java/org/apache/gobblin/service/ServiceConfigKeys.java
index 87bda430a..aeadf4dc6 100644
--- a/gobblin-api/src/main/java/org/apache/gobblin/service/ServiceConfigKeys.java
+++ b/gobblin-api/src/main/java/org/apache/gobblin/service/ServiceConfigKeys.java
@@ -204,4 +204,5 @@ public class ServiceConfigKeys {
public static final String DAG_PROCESSING_ENGINE_ENABLED = GOBBLIN_SERVICE_DAG_PROCESSING_ENGINE_PREFIX + "enabled";
public static final String NUM_DAG_PROC_THREADS_KEY = GOBBLIN_SERVICE_DAG_PROCESSING_ENGINE_PREFIX + "numThreads";
public static final Integer DEFAULT_NUM_DAG_PROC_THREADS = 3;
+ public static final String GOBBLIN_SERVICE_MULTI_ACTIVE_EXECUTION_ENABLED = GOBBLIN_SERVICE_PREFIX + "multiActiveExecutionEnabled";
}
diff --git a/gobblin-metrics-libs/gobblin-metrics/src/main/java/org/apache/gobblin/metrics/ServiceMetricNames.java b/gobblin-metrics-libs/gobblin-metrics/src/main/java/org/apache/gobblin/metrics/ServiceMetricNames.java
index 41e242b43..3af11c1f1 100644
--- a/gobblin-metrics-libs/gobblin-metrics/src/main/java/org/apache/gobblin/metrics/ServiceMetricNames.java
+++ b/gobblin-metrics-libs/gobblin-metrics/src/main/java/org/apache/gobblin/metrics/ServiceMetricNames.java
@@ -35,7 +35,8 @@ public class ServiceMetricNames {
public static final String FLOW_ORCHESTRATION_TIMER = GOBBLIN_SERVICE_PREFIX_WITH_DELIMITER + "flowOrchestration.time";
public static final String FLOW_ORCHESTRATION_DELAY = GOBBLIN_SERVICE_PREFIX_WITH_DELIMITER + "flowOrchestration.delay";
- // Flow Trigger Handler
+ // Flow Trigger Decorator
+ // TODO: change these metric variable and names after verifying the refactoring preserves existing functionality
public static final String FLOW_TRIGGER_HANDLER_PREFIX = GOBBLIN_SERVICE_PREFIX_WITH_DELIMITER + "flowTriggerHandler.";
public static final String GOBBLIN_FLOW_TRIGGER_HANDLER_NUM_FLOWS_SUBMITTED = FLOW_TRIGGER_HANDLER_PREFIX + "numFlowsSubmitted";
public static final String FLOW_TRIGGER_HANDLER_LEASE_OBTAINED_COUNT = FLOW_TRIGGER_HANDLER_PREFIX + "leaseObtained";
diff --git a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/DagActionStore.java b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/DagActionStore.java
index 4f3442597..562b3cb9e 100644
--- a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/DagActionStore.java
+++ b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/DagActionStore.java
@@ -26,7 +26,7 @@ import org.apache.gobblin.service.FlowId;
public interface DagActionStore {
- enum FlowActionType {
+ enum DagActionType {
KILL, // Kill invoked through API call
RESUME, // Resume flow invoked through API call
LAUNCH, // Launch new flow execution invoked adhoc or through scheduled trigger
@@ -40,7 +40,8 @@ public interface DagActionStore {
final String flowGroup;
final String flowName;
final String flowExecutionId;
- final FlowActionType flowActionType;
+ final String jobName;
+ final DagActionType dagActionType;
public FlowId getFlowId() {
return new FlowId().setFlowGroup(this.flowGroup).setFlowName(this.flowName);
@@ -51,30 +52,53 @@ public interface DagActionStore {
*/
public DagAction updateFlowExecutionId(long eventTimeMillis) {
return new DagAction(this.getFlowGroup(), this.getFlowName(),
- String.valueOf(eventTimeMillis), this.getFlowActionType());
+ String.valueOf(eventTimeMillis), this.getJobName(), this.getDagActionType());
}
}
/**
- * Check if an action exists in dagAction store by flow group, flow name and flow execution id.
+ * Check if an action exists in dagAction store by flow group, flow name, flow execution id, and job name.
* @param flowGroup flow group for the dag action
* @param flowName flow name for the dag action
* @param flowExecutionId flow execution for the dag action
- * @param flowActionType the value of the dag action
+ * @param jobName job name for the dag action
+ * @param dagActionType the value of the dag action
* @throws IOException
*/
- boolean exists(String flowGroup, String flowName, String flowExecutionId, FlowActionType flowActionType) throws IOException, SQLException;
+ boolean exists(String flowGroup, String flowName, String flowExecutionId, String jobName, DagActionType dagActionType) throws IOException, SQLException;
+
+ /**
+ * Check if an action exists in dagAction store by flow group, flow name, and flow execution id, it assumes jobName is
+ * empty ("").
+ * @param flowGroup flow group for the dag action
+ * @param flowName flow name for the dag action
+ * @param flowExecutionId flow execution for the dag action
+ * @param dagActionType the value of the dag action
+ * @throws IOException
+ */
+ boolean exists(String flowGroup, String flowName, String flowExecutionId, DagActionType dagActionType) throws IOException, SQLException;
/**
* Persist the dag action in {@link DagActionStore} for durability
* @param flowGroup flow group for the dag action
* @param flowName flow name for the dag action
* @param flowExecutionId flow execution for the dag action
- * @param flowActionType the value of the dag action
+ * @param jobName job name for the dag action
+ * @param dagActionType the value of the dag action
+ * @throws IOException
+ */
+ void addDagAction(String flowGroup, String flowName, String flowExecutionId, String jobName, DagActionType dagActionType) throws IOException;
+
+ /**
+ * Persist the dag action in {@link DagActionStore} for durability. This method assumes an empty jobName.
+ * @param flowGroup flow group for the dag action
+ * @param flowName flow name for the dag action
+ * @param flowExecutionId flow execution for the dag action
+ * @param dagActionType the value of the dag action
* @throws IOException
*/
- void addDagAction(String flowGroup, String flowName, String flowExecutionId, FlowActionType flowActionType) throws IOException;
+ void addDagAction(String flowGroup, String flowName, String flowExecutionId, DagActionType dagActionType) throws IOException;
/**
* delete the dag action from {@link DagActionStore}
diff --git a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/MultiActiveLeaseArbiter.java b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/MultiActiveLeaseArbiter.java
index 21befc26f..89df850be 100644
--- a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/MultiActiveLeaseArbiter.java
+++ b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/MultiActiveLeaseArbiter.java
@@ -20,6 +20,8 @@ package org.apache.gobblin.runtime.api;
import java.io.IOException;
import lombok.Data;
+import lombok.Getter;
+import lombok.AccessLevel;
/**
@@ -30,7 +32,7 @@ import lombok.Data;
* the attempt.
*
* At a high level the lease arbiter works as follows:
- * 1. Multiple participants independently learn of a flow action event to act upon
+ * 1. Multiple participants independently learn of a dag action event to act upon
* 2. Each participant attempts to acquire rights or `a lease` to be the sole participant acting on the event by
* calling the tryAcquireLease method below and receives the resulting status. The status indicates whether this
* participant has
@@ -45,27 +47,29 @@ import lombok.Data;
*/
public interface MultiActiveLeaseArbiter {
/**
- * This method attempts to insert an entry into store for a particular flow action event if one does not already
- * exist in the store for the flow action or has expired. Regardless of the outcome it also reads the lease
- * acquisition timestamp of the entry for that flow action event (it could have pre-existed in the table or been newly
+ * This method attempts to insert an entry into store for a particular dag action event if one does not already
+ * exist in the store for the dag action or has expired. Regardless of the outcome it also reads the lease
+ * acquisition timestamp of the entry for that dag action event (it could have pre-existed in the table or been newly
* added by the previous write). Based on the transaction results, it will return {@link LeaseAttemptStatus} to
* determine the next action.
- * @param flowAction uniquely identifies the flow and the present action upon it
- * @param eventTimeMillis is the time this flow action was triggered
- * @param isReminderEvent true if the flow action event we're checking on is a reminder event
- * @param adoptConsensusFlowExecutionId if true then replaces the flowAction flowExecutionId returned in
+ * @param dagAction uniquely identifies the flow and the present action upon it
+ * @param eventTimeMillis is the time this dag action was triggered
+ * @param isReminderEvent true if the dag action event we're checking on is a reminder event
+ * @param adoptConsensusFlowExecutionId if true then replaces the dagAction flowExecutionId returned in
* LeaseAttemptStatuses with the consensus eventTime
*
- * @return LeaseAttemptStatus
+ * @return LeaseAttemptStatus, containing a dag action that will have an updated flow execution id if `
+ * adoptConsensusFlowExecutionId` is true. The caller should use the newer version of the dag action to easily track
+ * the action moving forward.
* @throws IOException
*/
- LeaseAttemptStatus tryAcquireLease(DagActionStore.DagAction flowAction, long eventTimeMillis, boolean isReminderEvent,
+ LeaseAttemptStatus tryAcquireLease(DagActionStore.DagAction dagAction, long eventTimeMillis, boolean isReminderEvent,
boolean adoptConsensusFlowExecutionId)
throws IOException;
/**
* This method is used to indicate the owner of the lease has successfully completed required actions while holding
- * the lease of the flow action event. It marks the lease as "no longer leasing", if the eventTimeMillis and
+ * the lease of the dag action event. It marks the lease as "no longer leasing", if the eventTimeMillis and
* leaseAcquisitionTimeMillis values have not changed since this owner acquired the lease (indicating the lease did
* not expire).
* @return true if successfully updated, indicating no further actions need to be taken regarding this event.
@@ -76,40 +80,62 @@ public interface MultiActiveLeaseArbiter {
/*
Class used to encapsulate status of lease acquisition attempt and derivations should contain information specific to
- the status that results.
+ the status that results. The #getDagAction and #getMinimumLingerDurationMillis are meant to be overriden and used by
+ relevant derived classes.
*/
- abstract class LeaseAttemptStatus {}
+ abstract class LeaseAttemptStatus {
+ public DagActionStore.DagAction getDagAction() {
+ return null;
+ }
+
+ public long getMinimumLingerDurationMillis() {
+ return 0;
+ }
+ }
class NoLongerLeasingStatus extends LeaseAttemptStatus {}
/*
- The participant calling this method acquired the lease for the event in question. `Flow action`'s flow execution id
+ The participant calling this method acquired the lease for the event in question. `Dag action`'s flow execution id
is the timestamp associated with the lease and the time the caller obtained the lease is stored within the
- `leaseAcquisitionTimestamp` field.
+ `leaseAcquisitionTimestamp` field. The `multiActiveLeaseArbiter` reference is used to recordLeaseSuccess for the
+ current LeaseObtainedStatus via the completeLease method from a caller without access to the {@link MultiActiveLeaseArbiter}.
*/
@Data
class LeaseObtainedStatus extends LeaseAttemptStatus {
- private final DagActionStore.DagAction flowAction;
+ private final DagActionStore.DagAction dagAction;
private final long leaseAcquisitionTimestamp;
+ private final long minimumLingerDurationMillis;
+ @Getter(AccessLevel.NONE)
+ private final MultiActiveLeaseArbiter multiActiveLeaseArbiter;
/**
* @return event time in millis since epoch for the event of this lease acquisition
*/
public long getEventTimeMillis() {
- return Long.parseLong(flowAction.getFlowExecutionId());
+ return Long.parseLong(dagAction.getFlowExecutionId());
+ }
+
+ /**
+ * Completes the lease referenced by this status object if it has not expired.
+ * @return true if able to complete lease, false otherwise.
+ * @throws IOException
+ */
+ public boolean completeLease() throws IOException {
+ return multiActiveLeaseArbiter.recordLeaseSuccess(this);
}
}
/*
- This flow action event already has a valid lease owned by another participant.
- `Flow action`'s flow execution id is the timestamp the lease is associated with, however the flow action event it
+ This dag action event already has a valid lease owned by another participant.
+ `Dag action`'s flow execution id is the timestamp the lease is associated with, however the dag action event it
corresponds to may be a different and distinct occurrence of the same event.
`minimumLingerDurationMillis` is the minimum amount of time to wait before this participant should return to check if
the lease has completed or expired
*/
@Data
class LeasedToAnotherStatus extends LeaseAttemptStatus {
- private final DagActionStore.DagAction flowAction;
+ private final DagActionStore.DagAction dagAction;
private final long minimumLingerDurationMillis;
/**
@@ -117,7 +143,7 @@ public interface MultiActiveLeaseArbiter {
* @return
*/
public long getEventTimeMillis() {
- return Long.parseLong(flowAction.getFlowExecutionId());
+ return Long.parseLong(dagAction.getFlowExecutionId());
}
}
}
diff --git a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/MysqlMultiActiveLeaseArbiter.java b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/MysqlMultiActiveLeaseArbiter.java
index 4f86c85eb..88915b368 100644
--- a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/MysqlMultiActiveLeaseArbiter.java
+++ b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/MysqlMultiActiveLeaseArbiter.java
@@ -43,11 +43,11 @@ import org.apache.gobblin.util.DBStatementExecutor;
/**
* MySQL based implementation of the {@link MultiActiveLeaseArbiter} which uses a MySQL store to resolve ownership of
- * a flow event amongst multiple competing participants. A MySQL table is used to store flow identifying information as
- * well as the flow action associated with it. It uses two additional values of the `event_timestamp` and
+ * a dag action event amongst multiple competing participants. A MySQL table is used to store flow and job identifying
+ * information as well as the dag action associated with it. It uses two additional values of the `event_timestamp` and
* `lease_acquisition_timestamp` to indicate an active lease, expired lease, and state of no longer leasing. The table
* schema is as follows:
- * [flow_group | flow_name | flow_action | event_timestamp | lease_acquisition_timestamp]
+ * [flow_group | flow_name | job_name | dag_action | event_timestamp | lease_acquisition_timestamp]
* (----------------------primary key------------------------)
* We also maintain another table in the database with two constants that allow us to coordinate between participants
* and ensure they are using the same values to base their coordination off of.
@@ -56,24 +56,24 @@ import org.apache.gobblin.util.DBStatementExecutor;
* `linger` - minimum time to occur before another host may attempt a lease on a flow event. It should be much greater
* than epsilon and encapsulate executor communication latency including retry attempts
*
- * The `event_timestamp` is the time of the flow_action event request.
+ * The `event_timestamp` is the time of the dag_action event request.
* --- Database event_timestamp laundering ---
- * We only use the participant's local event_timestamp internally to identify the particular flow_action event, but
+ * We only use the participant's local event_timestamp internally to identify the particular dag_action event, but
* after interacting with the database utilize the CURRENT_TIMESTAMP of the database to insert or keep
* track of our event, "laundering" or replacing the local timestamp with the database one. This is to avoid any
* discrepancies due to clock drift between participants as well as variation in local time and database time for
* future comparisons.
* --- Event consolidation ---
- * Note that for the sake of simplification, we only allow one event associated with a particular flow's flow_action
+ * Note that for the sake of simplification, we only allow one event associated with a particular flow's dag_action
* (ie: only one LAUNCH for example of flow FOO, but there can be a LAUNCH, KILL, & RESUME for flow FOO at once) during
- * the time it takes to execute the flow action. In most cases, the execution time should be so negligible that this
- * event consolidation of duplicate flow action requests is not noticed and even during executor downtime this behavior
+ * the time it takes to execute the dag action. In most cases, the execution time should be so negligible that this
+ * event consolidation of duplicate dag action requests is not noticed and even during executor downtime this behavior
* is acceptable as the user generally expects a timely execution of the most recent request rather than one execution
* per request.
*
- * The `lease_acquisition_timestamp` is the time a host acquired ownership of this flow action, and it is valid for
+ * The `lease_acquisition_timestamp` is the time a host acquired ownership of this dag action, and it is valid for
* `linger` period of time after which it expires and any host can re-attempt ownership. In most cases, the original
- * host should actually complete its work while having the lease and then mark the flow action as NULL to indicate no
+ * host should actually complete its work while having the lease and then mark the dag action as NULL to indicate no
* further leasing should be done for the event.
*/
@Slf4j
@@ -108,10 +108,11 @@ public class MysqlMultiActiveLeaseArbiter implements MultiActiveLeaseArbiter {
*/
private static final String CREATE_LEASE_ARBITER_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
+ "flow_group varchar(" + ServiceConfigKeys.MAX_FLOW_GROUP_LENGTH + ") NOT NULL, flow_name varchar("
- + ServiceConfigKeys.MAX_FLOW_GROUP_LENGTH + ") NOT NULL, " + " flow_action varchar(100) NOT NULL, "
+ + ServiceConfigKeys.MAX_FLOW_GROUP_LENGTH + ") NOT NULL, " + "job_name varchar("
+ + ServiceConfigKeys.MAX_FLOW_GROUP_LENGTH + ") NOT NULL, dag_action varchar(100) NOT NULL, "
+ "event_timestamp TIMESTAMP(3) NOT NULL, "
+ "lease_acquisition_timestamp TIMESTAMP(3) NULL, "
- + "PRIMARY KEY (flow_group,flow_name,flow_action))";
+ + "PRIMARY KEY (flow_group,flow_name,job_name,dag_action))";
// Deletes rows older than retention time period regardless of lease status as they should all be invalid or completed
// since retention >> linger
private static final String LEASE_ARBITER_TABLE_RETENTION_STATEMENT = "DELETE FROM %s WHERE event_timestamp < "
@@ -121,7 +122,7 @@ public class MysqlMultiActiveLeaseArbiter implements MultiActiveLeaseArbiter {
// Only insert epsilon and linger values from config if this table does not contain a pre-existing values already.
private static final String UPSERT_CONSTANTS_TABLE_STATEMENT = "INSERT INTO %s (primary_key, epsilon, linger) "
+ "VALUES(1, ?, ?) ON DUPLICATE KEY UPDATE epsilon=VALUES(epsilon), linger=VALUES(linger)";
- protected static final String WHERE_CLAUSE_TO_MATCH_KEY = "WHERE flow_group=? AND flow_name=? AND flow_action=?";
+ protected static final String WHERE_CLAUSE_TO_MATCH_KEY = "WHERE flow_group=? AND flow_name=? AND job_name=? AND dag_action=?";
protected static final String WHERE_CLAUSE_TO_MATCH_ROW = WHERE_CLAUSE_TO_MATCH_KEY
+ " AND event_timestamp=CONVERT_TZ(?, '+00:00', @@session.time_zone)"
+ " AND lease_acquisition_timestamp=CONVERT_TZ(?, '+00:00', @@session.time_zone)";
@@ -155,8 +156,8 @@ public class MysqlMultiActiveLeaseArbiter implements MultiActiveLeaseArbiter {
+ WHERE_CLAUSE_TO_MATCH_KEY;
// Insert or update row to acquire lease if values have not changed since the previous read
// Need to define three separate statements to handle cases where row does not exist or has null values to check
- protected static final String ACQUIRE_LEASE_IF_NEW_ROW_STATEMENT = "INSERT INTO %s (flow_group, flow_name, "
- + "flow_action, event_timestamp, lease_acquisition_timestamp) VALUES(?, ?, ?, CURRENT_TIMESTAMP(3), "
+ protected static final String ACQUIRE_LEASE_IF_NEW_ROW_STATEMENT = "INSERT INTO %s (flow_group, flow_name, job_name, "
+ + "dag_action, event_timestamp, lease_acquisition_timestamp) VALUES(?, ?, ?, ?, CURRENT_TIMESTAMP(3), "
+ "CURRENT_TIMESTAMP(3))";
protected static final String CONDITIONALLY_ACQUIRE_LEASE_IF_FINISHED_LEASING_STATEMENT = "UPDATE %s "
+ "SET event_timestamp=CURRENT_TIMESTAMP(3), lease_acquisition_timestamp=CURRENT_TIMESTAMP(3) "
@@ -180,6 +181,7 @@ public class MysqlMultiActiveLeaseArbiter implements MultiActiveLeaseArbiter {
+ "before all properties", ConfigurationKeys.MYSQL_LEASE_ARBITER_PREFIX));
}
+ // TODO: create two tables or take in table name as parameter to have scheduler and executor table
this.leaseArbiterTableName = ConfigUtils.getString(config, ConfigurationKeys.SCHEDULER_LEASE_DETERMINATION_STORE_DB_TABLE_KEY,
ConfigurationKeys.DEFAULT_SCHEDULER_LEASE_DETERMINATION_STORE_DB_TABLE);
this.constantsTableName = ConfigUtils.getString(config, ConfigurationKeys.MULTI_ACTIVE_SCHEDULER_CONSTANTS_DB_TABLE_KEY,
@@ -237,19 +239,19 @@ public class MysqlMultiActiveLeaseArbiter implements MultiActiveLeaseArbiter {
}
@Override
- public LeaseAttemptStatus tryAcquireLease(DagActionStore.DagAction flowAction, long eventTimeMillis,
+ public LeaseAttemptStatus tryAcquireLease(DagActionStore.DagAction dagAction, long eventTimeMillis,
boolean isReminderEvent, boolean adoptConsensusFlowExecutionId) throws IOException {
log.info("Multi-active scheduler about to handle trigger event: [{}, is: {}, triggerEventTimestamp: {}]",
- flowAction, isReminderEvent ? "reminder" : "original", eventTimeMillis);
- // Query lease arbiter table about this flow action
- Optional<GetEventInfoResult> getResult = getExistingEventInfo(flowAction, isReminderEvent, eventTimeMillis);
+ dagAction, isReminderEvent ? "reminder" : "original", eventTimeMillis);
+ // Query lease arbiter table about this dag action
+ Optional<GetEventInfoResult> getResult = getExistingEventInfo(dagAction, isReminderEvent, eventTimeMillis);
try {
if (!getResult.isPresent()) {
- log.debug("tryAcquireLease for [{}, is; {}, eventTimestamp: {}] - CASE 1: no existing row for this flow action,"
- + " then go ahead and insert", flowAction, isReminderEvent ? "reminder" : "original", eventTimeMillis);
- int numRowsUpdated = attemptLeaseIfNewRow(flowAction);
- return evaluateStatusAfterLeaseAttempt(numRowsUpdated, flowAction, Optional.empty(),
+ log.debug("tryAcquireLease for [{}, is; {}, eventTimestamp: {}] - CASE 1: no existing row for this dag action,"
+ + " then go ahead and insert", dagAction, isReminderEvent ? "reminder" : "original", eventTimeMillis);
+ int numRowsUpdated = attemptLeaseIfNewRow(dagAction);
+ return evaluateStatusAfterLeaseAttempt(numRowsUpdated, dagAction, Optional.empty(),
isReminderEvent, adoptConsensusFlowExecutionId);
}
@@ -267,7 +269,7 @@ public class MysqlMultiActiveLeaseArbiter implements MultiActiveLeaseArbiter {
if (isReminderEvent) {
if (eventTimeMillis < dbEventTimestamp.getTime()) {
log.debug("tryAcquireLease for [{}, is: {}, eventTimestamp: {}] - dbEventTimeMillis: {} - A new event trigger "
- + "is being worked on, so this older reminder will be dropped.", flowAction,
+ + "is being worked on, so this older reminder will be dropped.", dagAction,
isReminderEvent ? "reminder" : "original", eventTimeMillis, dbEventTimestamp);
return new NoLongerLeasingStatus();
}
@@ -275,18 +277,18 @@ public class MysqlMultiActiveLeaseArbiter implements MultiActiveLeaseArbiter {
// TODO: emit metric here to capture this unexpected behavior
log.warn("tryAcquireLease for [{}, is: {}, eventTimestamp: {}] - dbEventTimeMillis: {} - Severe constraint "
+ "violation encountered: a reminder event newer than db event was found when db laundering should "
- + "ensure monotonically increasing laundered event times.", flowAction,
+ + "ensure monotonically increasing laundered event times.", dagAction,
isReminderEvent ? "reminder" : "original", eventTimeMillis, dbEventTimestamp.getTime());
}
if (eventTimeMillis == dbEventTimestamp.getTime()) {
log.debug("tryAcquireLease for [{}, is: {}, eventTimestamp: {}] - dbEventTimeMillis: {} - Reminder event time "
- + "is the same as db event.", flowAction, isReminderEvent ? "reminder" : "original",
+ + "is the same as db event.", dagAction, isReminderEvent ? "reminder" : "original",
eventTimeMillis, dbEventTimestamp);
}
}
log.info("Multi-active arbiter replacing local trigger event timestamp [{}, is: {}, triggerEventTimestamp: {}] "
- + "with database eventTimestamp {} (in epoch-millis)", flowAction, isReminderEvent ? "reminder" : "original",
+ + "with database eventTimestamp {} (in epoch-millis)", dagAction, isReminderEvent ? "reminder" : "original",
eventTimeMillis, dbCurrentTimestamp.getTime());
/* Note that we use `adoptConsensusFlowExecutionId` parameter's value to determine whether we should use the db
@@ -296,48 +298,48 @@ public class MysqlMultiActiveLeaseArbiter implements MultiActiveLeaseArbiter {
// Lease is valid
if (leaseValidityStatus == 1) {
if (isWithinEpsilon) {
- DagActionStore.DagAction updatedFlowAction =
- adoptConsensusFlowExecutionId ? flowAction.updateFlowExecutionId(dbEventTimestamp.getTime()) : flowAction;
+ DagActionStore.DagAction updatedDagAction =
+ adoptConsensusFlowExecutionId ? dagAction.updateFlowExecutionId(dbEventTimestamp.getTime()) : dagAction;
log.debug("tryAcquireLease for [{}, is: {}, eventTimestamp: {}] - CASE 2: Same event, lease is valid",
- updatedFlowAction, isReminderEvent ? "reminder" : "original", dbCurrentTimestamp.getTime());
+ updatedDagAction, isReminderEvent ? "reminder" : "original", dbCurrentTimestamp.getTime());
// Utilize db timestamp for reminder
- return new LeasedToAnotherStatus(updatedFlowAction,
+ return new LeasedToAnotherStatus(updatedDagAction,
dbLeaseAcquisitionTimestamp.getTime() + dbLinger - dbCurrentTimestamp.getTime());
}
- DagActionStore.DagAction updatedFlowAction =
- adoptConsensusFlowExecutionId ? flowAction.updateFlowExecutionId(dbCurrentTimestamp.getTime()) : flowAction;
+ DagActionStore.DagAction updatedDagAction =
+ adoptConsensusFlowExecutionId ? dagAction.updateFlowExecutionId(dbCurrentTimestamp.getTime()) : dagAction;
log.debug("tryAcquireLease for [{}, is: {}, eventTimestamp: {}] - CASE 3: Distinct event, lease is valid",
- updatedFlowAction, isReminderEvent ? "reminder" : "original", dbCurrentTimestamp.getTime());
+ updatedDagAction, isReminderEvent ? "reminder" : "original", dbCurrentTimestamp.getTime());
// Utilize db lease acquisition timestamp for wait time
- return new LeasedToAnotherStatus(updatedFlowAction,
+ return new LeasedToAnotherStatus(updatedDagAction,
dbLeaseAcquisitionTimestamp.getTime() + dbLinger - dbCurrentTimestamp.getTime());
} // Lease is invalid
else if (leaseValidityStatus == 2) {
log.debug("tryAcquireLease for [{}, is: {}, eventTimestamp: {}] - CASE 4: Lease is out of date (regardless of "
- + "whether same or distinct event)", flowAction, isReminderEvent ? "reminder" : "original",
+ + "whether same or distinct event)", dagAction, isReminderEvent ? "reminder" : "original",
dbCurrentTimestamp.getTime());
if (isWithinEpsilon && !isReminderEvent) {
- log.warn("Lease should not be out of date for the same trigger event since epsilon << linger for flowAction"
- + " {}, db eventTimestamp {}, db leaseAcquisitionTimestamp {}, linger {}", flowAction,
+ log.warn("Lease should not be out of date for the same trigger event since epsilon << linger for dagAction"
+ + " {}, db eventTimestamp {}, db leaseAcquisitionTimestamp {}, linger {}", dagAction,
dbEventTimestamp, dbLeaseAcquisitionTimestamp, dbLinger);
}
// Use our event to acquire lease, check for previous db eventTimestamp and leaseAcquisitionTimestamp
- int numRowsUpdated = attemptLeaseIfExistingRow(thisTableAcquireLeaseIfMatchingAllStatement, flowAction,
+ int numRowsUpdated = attemptLeaseIfExistingRow(thisTableAcquireLeaseIfMatchingAllStatement, dagAction,
true,true, dbEventTimestamp, dbLeaseAcquisitionTimestamp);
- return evaluateStatusAfterLeaseAttempt(numRowsUpdated, flowAction, Optional.of(dbCurrentTimestamp),
+ return evaluateStatusAfterLeaseAttempt(numRowsUpdated, dagAction, Optional.of(dbCurrentTimestamp),
isReminderEvent, adoptConsensusFlowExecutionId);
} // No longer leasing this event
if (isWithinEpsilon) {
log.debug("tryAcquireLease for [{}, is: {}, eventTimestamp: {}] - CASE 5: Same event, no longer leasing event"
- + " in db", flowAction, isReminderEvent ? "reminder" : "original", dbCurrentTimestamp.getTime());
+ + " in db", dagAction, isReminderEvent ? "reminder" : "original", dbCurrentTimestamp.getTime());
return new NoLongerLeasingStatus();
}
log.debug("tryAcquireLease for [{}, is: {}, eventTimestamp: {}] - CASE 6: Distinct event, no longer leasing "
- + "event in db", flowAction, isReminderEvent ? "reminder" : "original", dbCurrentTimestamp.getTime());
+ + "event in db", dagAction, isReminderEvent ? "reminder" : "original", dbCurrentTimestamp.getTime());
// Use our event to acquire lease, check for previous db eventTimestamp and NULL leaseAcquisitionTimestamp
- int numRowsUpdated = attemptLeaseIfExistingRow(thisTableAcquireLeaseIfFinishedStatement, flowAction,
+ int numRowsUpdated = attemptLeaseIfExistingRow(thisTableAcquireLeaseIfFinishedStatement, dagAction,
true, false, dbEventTimestamp, null);
- return evaluateStatusAfterLeaseAttempt(numRowsUpdated, flowAction, Optional.of(dbCurrentTimestamp),
+ return evaluateStatusAfterLeaseAttempt(numRowsUpdated, dagAction, Optional.of(dbCurrentTimestamp),
isReminderEvent, adoptConsensusFlowExecutionId);
} catch (SQLException e) {
throw new RuntimeException(e);
@@ -345,9 +347,9 @@ public class MysqlMultiActiveLeaseArbiter implements MultiActiveLeaseArbiter {
}
/**
- * Checks leaseArbiterTable for an existing entry for this flow action and event time
+ * Checks leaseArbiterTable for an existing entry for this dag action and event time
*/
- protected Optional<GetEventInfoResult> getExistingEventInfo(DagActionStore.DagAction flowAction,
+ protected Optional<GetEventInfoResult> getExistingEventInfo(DagActionStore.DagAction dagAction,
boolean isReminderEvent, long eventTimeMillis) throws IOException {
return dbStatementExecutor.withPreparedStatement(isReminderEvent ? thisTableGetInfoStatementForReminder : thisTableGetInfoStatement,
getInfoStatement -> {
@@ -355,9 +357,10 @@ public class MysqlMultiActiveLeaseArbiter implements MultiActiveLeaseArbiter {
if (isReminderEvent) {
getInfoStatement.setTimestamp(++i, new Timestamp(eventTimeMillis), UTC_CAL.get());
}
- getInfoStatement.setString(++i, flowAction.getFlowGroup());
- getInfoStatement.setString(++i, flowAction.getFlowName());
- getInfoStatement.setString(++i, flowAction.getFlowActionType().toString());
+ getInfoStatement.setString(++i, dagAction.getFlowGroup());
+ getInfoStatement.setString(++i, dagAction.getFlowName());
+ getInfoStatement.setString(++i, dagAction.getJobName());
+ getInfoStatement.setString(++i, dagAction.getDagActionType().toString());
ResultSet resultSet = getInfoStatement.executeQuery();
try {
if (!resultSet.next()) {
@@ -397,16 +400,16 @@ public class MysqlMultiActiveLeaseArbiter implements MultiActiveLeaseArbiter {
}
/**
- * Called by participant to try to acquire lease for a flow action that does not have an attempt in progress or in
+ * Called by participant to try to acquire lease for a dag action that does not have an attempt in progress or in
* near past for it.
* @return int corresponding to number of rows updated by INSERT statement to acquire lease
*/
- protected int attemptLeaseIfNewRow(DagActionStore.DagAction flowAction) throws IOException {
+ protected int attemptLeaseIfNewRow(DagActionStore.DagAction dagAction) throws IOException {
String formattedAcquireLeaseNewRowStatement =
String.format(ACQUIRE_LEASE_IF_NEW_ROW_STATEMENT, this.leaseArbiterTableName);
return dbStatementExecutor.withPreparedStatement(formattedAcquireLeaseNewRowStatement,
insertStatement -> {
- completeInsertPreparedStatement(insertStatement, flowAction);
+ completeInsertPreparedStatement(insertStatement, dagAction);
try {
return insertStatement.executeUpdate();
} catch (SQLIntegrityConstraintViolationException e) {
@@ -419,29 +422,29 @@ public class MysqlMultiActiveLeaseArbiter implements MultiActiveLeaseArbiter {
}
/**
- * Called by participant to try to acquire lease for a flow action that has an existing, completed, or expired lease
- * attempt for the flow action in the table.
+ * Called by participant to try to acquire lease for a dag action that has an existing, completed, or expired lease
+ * attempt for the dag action in the table.
* @return int corresponding to number of rows updated by INSERT statement to acquire lease
*/
- protected int attemptLeaseIfExistingRow(String acquireLeaseStatement, DagActionStore.DagAction flowAction,
+ protected int attemptLeaseIfExistingRow(String acquireLeaseStatement, DagActionStore.DagAction dagAction,
boolean needEventTimeCheck, boolean needLeaseAcquisition, Timestamp dbEventTimestamp,
Timestamp dbLeaseAcquisitionTimestamp) throws IOException {
return dbStatementExecutor.withPreparedStatement(acquireLeaseStatement,
insertStatement -> {
- completeUpdatePreparedStatement(insertStatement, flowAction, needEventTimeCheck, needLeaseAcquisition,
+ completeUpdatePreparedStatement(insertStatement, dagAction, needEventTimeCheck, needLeaseAcquisition,
dbEventTimestamp, dbLeaseAcquisitionTimestamp);
return insertStatement.executeUpdate();
}, true);
}
/**
- * Checks leaseArbiter table for a row corresponding to this flow action to determine if the lease acquisition attempt
+ * Checks leaseArbiter table for a row corresponding to this dag action to determine if the lease acquisition attempt
* was successful or not.
*/
- protected SelectInfoResult getRowInfo(DagActionStore.DagAction flowAction) throws IOException {
+ protected SelectInfoResult getRowInfo(DagActionStore.DagAction dagAction) throws IOException {
return dbStatementExecutor.withPreparedStatement(thisTableSelectAfterInsertStatement,
selectStatement -> {
- completeWhereClauseMatchingKeyPreparedStatement(selectStatement, flowAction);
+ completeWhereClauseMatchingKeyPreparedStatement(selectStatement, dagAction);
ResultSet resultSet = selectStatement.executeQuery();
try {
return createSelectInfoResult(resultSet);
@@ -489,64 +492,68 @@ public class MysqlMultiActiveLeaseArbiter implements MultiActiveLeaseArbiter {
* @throws IOException
*/
protected LeaseAttemptStatus evaluateStatusAfterLeaseAttempt(int numRowsUpdated,
- DagActionStore.DagAction flowAction, Optional<Timestamp> dbCurrentTimestamp, boolean isReminderEvent,
+ DagActionStore.DagAction dagAction, Optional<Timestamp> dbCurrentTimestamp, boolean isReminderEvent,
boolean adoptConsensusFlowExecutionId)
throws SQLException, IOException {
// Fetch values in row after attempted insert
- SelectInfoResult selectInfoResult = getRowInfo(flowAction);
+ SelectInfoResult selectInfoResult = getRowInfo(dagAction);
// Another participant won the lease in between
if (!selectInfoResult.getLeaseAcquisitionTimeMillis().isPresent()) {
return new NoLongerLeasingStatus();
}
- DagActionStore.DagAction updatedFlowAction =
- adoptConsensusFlowExecutionId ? flowAction.updateFlowExecutionId(selectInfoResult.eventTimeMillis) : flowAction;
+ DagActionStore.DagAction updatedDagAction =
+ adoptConsensusFlowExecutionId ? dagAction.updateFlowExecutionId(selectInfoResult.eventTimeMillis) : dagAction;
+ // If no db current timestamp is present, then use the full db linger value for duration
+ long minimumLingerDurationMillis = dbCurrentTimestamp.isPresent() ?
+ selectInfoResult.getLeaseAcquisitionTimeMillis().get() + selectInfoResult.getDbLinger()
+ - dbCurrentTimestamp.get().getTime() : selectInfoResult.getDbLinger();
if (numRowsUpdated == 1) {
- log.info("Obtained lease for [{}, is: {}, eventTimestamp: {}] successfully!", updatedFlowAction,
+ log.info("Obtained lease for [{}, is: {}, eventTimestamp: {}] successfully!", updatedDagAction,
isReminderEvent ? "reminder" : "original", selectInfoResult.eventTimeMillis);
- return new LeaseObtainedStatus(updatedFlowAction, selectInfoResult.getLeaseAcquisitionTimeMillis().get());
+ return new LeaseObtainedStatus(updatedDagAction, selectInfoResult.getLeaseAcquisitionTimeMillis().get(), minimumLingerDurationMillis, this);
}
log.info("Another participant acquired lease in between for [{}, is: {}, eventTimestamp: {}] - num rows updated: {}",
- updatedFlowAction, isReminderEvent ? "reminder" : "original", selectInfoResult.eventTimeMillis, numRowsUpdated);
+ updatedDagAction, isReminderEvent ? "reminder" : "original", selectInfoResult.eventTimeMillis, numRowsUpdated);
// Another participant acquired lease in between
- return new LeasedToAnotherStatus(updatedFlowAction,
- selectInfoResult.getLeaseAcquisitionTimeMillis().get() + selectInfoResult.getDbLinger()
- - (dbCurrentTimestamp.isPresent() ? dbCurrentTimestamp.get().getTime() : System.currentTimeMillis()));
+ return new LeasedToAnotherStatus(updatedDagAction, minimumLingerDurationMillis);
}
/**
- * Complete the INSERT statement for a new flow action lease where the flow action is not present in the table
+ * Complete the INSERT statement for a new dag action lease where the dag action is not present in the table
* @param statement
- * @param flowAction
+ * @param dagAction
* @throws SQLException
*/
protected static void completeInsertPreparedStatement(PreparedStatement statement,
- DagActionStore.DagAction flowAction) throws SQLException {
+ DagActionStore.DagAction dagAction) throws SQLException {
int i = 0;
// Values to set in new row
- statement.setString(++i, flowAction.getFlowGroup());
- statement.setString(++i, flowAction.getFlowName());
- statement.setString(++i, flowAction.getFlowActionType().toString());
+ statement.setString(++i, dagAction.getFlowGroup());
+ statement.setString(++i, dagAction.getFlowName());
+ statement.setString(++i, dagAction.getJobName());
+ statement.setString(++i, dagAction.getDagActionType().toString());
}
/**
- * Complete the WHERE clause to match a flow action in a select statement
+ * Complete the WHERE clause to match a dag action in a select statement
* @param statement
- * @param flowAction
+ * @param dagAction
* @throws SQLException
*/
protected static void completeWhereClauseMatchingKeyPreparedStatement(PreparedStatement statement,
- DagActionStore.DagAction flowAction) throws SQLException {
+ DagActionStore.DagAction dagAction) throws SQLException {
int i = 0;
- statement.setString(++i, flowAction.getFlowGroup());
- statement.setString(++i, flowAction.getFlowName());
- statement.setString(++i, flowAction.getFlowActionType().toString());
+ statement.setString(++i, dagAction.getFlowGroup());
+ statement.setString(++i, dagAction.getFlowName());
+ statement.setString(++i, dagAction.getJobName());
+ statement.setString(++i, dagAction.getDagActionType().toString());
}
/**
- * Complete the UPDATE prepared statements for a flow action that already exists in the table that needs to be
+ * Complete the UPDATE prepared statements for a dag action that already exists in the table that needs to be
* updated.
* @param statement
- * @param flowAction
+ * @param dagAction
* @param needEventTimeCheck true if need to compare `originalEventTimestamp` with db event_timestamp
* @param needLeaseAcquisitionTimeCheck true if need to compare `originalLeaseAcquisitionTimestamp` with db one
* @param originalEventTimestamp value to compare to db one, null if not needed
@@ -554,13 +561,14 @@ public class MysqlMultiActiveLeaseArbiter implements MultiActiveLeaseArbiter {
* @throws SQLException
*/
protected static void completeUpdatePreparedStatement(PreparedStatement statement,
- DagActionStore.DagAction flowAction, boolean needEventTimeCheck, boolean needLeaseAcquisitionTimeCheck,
+ DagActionStore.DagAction dagAction, boolean needEventTimeCheck, boolean needLeaseAcquisitionTimeCheck,
Timestamp originalEventTimestamp, Timestamp originalLeaseAcquisitionTimestamp) throws SQLException {
int i = 0;
// Values to check if existing row matches previous read
- statement.setString(++i, flowAction.getFlowGroup());
- statement.setString(++i, flowAction.getFlowName());
- statement.setString(++i, flowAction.getFlowActionType().toString());
+ statement.setString(++i, dagAction.getFlowGroup());
+ statement.setString(++i, dagAction.getFlowName());
+ statement.setString(++i, dagAction.getJobName());
+ statement.setString(++i, dagAction.getDagActionType().toString());
// Values that may be needed depending on the insert statement
if (needEventTimeCheck) {
statement.setTimestamp(++i, originalEventTimestamp, UTC_CAL.get());
@@ -573,37 +581,35 @@ public class MysqlMultiActiveLeaseArbiter implements MultiActiveLeaseArbiter {
@Override
public boolean recordLeaseSuccess(LeaseObtainedStatus status)
throws IOException {
- DagActionStore.DagAction flowAction = status.getFlowAction();
- String flowGroup = flowAction.getFlowGroup();
- String flowName = flowAction.getFlowName();
- DagActionStore.FlowActionType flowActionType = flowAction.getFlowActionType();
+ DagActionStore.DagAction dagAction = status.getDagAction();
return dbStatementExecutor.withPreparedStatement(String.format(CONDITIONALLY_COMPLETE_LEASE_STATEMENT, leaseArbiterTableName),
updateStatement -> {
int i = 0;
- updateStatement.setString(++i, flowGroup);
- updateStatement.setString(++i, flowName);
- updateStatement.setString(++i, flowActionType.toString());
+ updateStatement.setString(++i, dagAction.getFlowGroup());
+ updateStatement.setString(++i, dagAction.getFlowName());
+ updateStatement.setString(++i, dagAction.getJobName());
+ updateStatement.setString(++i, dagAction.getDagActionType().toString());
updateStatement.setTimestamp(++i, new Timestamp(status.getEventTimeMillis()), UTC_CAL.get());
updateStatement.setTimestamp(++i, new Timestamp(status.getLeaseAcquisitionTimestamp()), UTC_CAL.get());
int numRowsUpdated = updateStatement.executeUpdate();
if (numRowsUpdated == 0) {
log.info("Multi-active lease arbiter lease attempt: [{}, eventTimestamp: {}] - FAILED to complete because "
- + "lease expired or event cleaned up before host completed required actions", flowAction,
+ + "lease expired or event cleaned up before host completed required actions", dagAction,
status.getEventTimeMillis());
return false;
}
if( numRowsUpdated == 1) {
log.info("Multi-active lease arbiter lease attempt: [{}, eventTimestamp: {}] - COMPLETED, no longer leasing"
- + " this event after this.", flowAction, status.getEventTimeMillis());
+ + " this event after this.", dagAction, status.getEventTimeMillis());
return true;
};
throw new IOException(String.format("Attempt to complete lease use: [%s, eventTimestamp: %s] - updated more "
- + "rows than expected", flowAction, status.getEventTimeMillis()));
+ + "rows than expected", dagAction, status.getEventTimeMillis()));
}, true);
}
/**
- * DTO for arbiter's current lease state for a FlowActionEvent
+ * DTO for arbiter's current lease state for a DagActionEvent
*/
@Data
static class GetEventInfoResult {
diff --git a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/dag_action_store/MysqlDagActionStore.java b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/dag_action_store/MysqlDagActionStore.java
index 894d0a300..d02f5333c 100644
--- a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/dag_action_store/MysqlDagActionStore.java
+++ b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/dag_action_store/MysqlDagActionStore.java
@@ -46,24 +46,26 @@ import org.apache.gobblin.util.DBStatementExecutor;
public class MysqlDagActionStore implements DagActionStore {
public static final String CONFIG_PREFIX = "MysqlDagActionStore";
+ public static final String NO_JOB_NAME_DEFAULT = "";
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 dag_action = ?)";
+ 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, dag_action) "
- + "VALUES (?, ?, ?, ?)";
- private static final String DELETE_STATEMENT = "DELETE FROM %s WHERE flow_group = ? AND flow_name =? AND flow_execution_id = ? AND dag_action = ?";
- private static final String GET_STATEMENT = "SELECT flow_group, flow_name, flow_execution_id, dag_action FROM %s WHERE flow_group = ? AND flow_name =? AND flow_execution_id = ? AND dag_action = ?";
- private static final String GET_ALL_STATEMENT = "SELECT flow_group, flow_name, flow_execution_id, dag_action FROM %s";
+ 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 = ?";
+ private static final String GET_ALL_STATEMENT = "SELECT flow_group, flow_name, flow_execution_id, job_name, dag_action FROM %s";
private static final String CREATE_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s (" +
"flow_group varchar(" + ServiceConfigKeys.MAX_FLOW_GROUP_LENGTH + ") NOT NULL, flow_name varchar(" + ServiceConfigKeys.MAX_FLOW_GROUP_LENGTH + ") NOT NULL, "
+ "flow_execution_id varchar(" + ServiceConfigKeys.MAX_FLOW_EXECUTION_ID_LENGTH + ") NOT NULL, "
- + "dag_action varchar(100) NOT NULL, modified_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP on update CURRENT_TIMESTAMP NOT NULL, "
- + "PRIMARY KEY (flow_group,flow_name,flow_execution_id, dag_action))";
+ + "job_name varchar(" + ServiceConfigKeys.MAX_FLOW_EXECUTION_ID_LENGTH + ") NOT NULL, "
+ + "dag_action varchar(50) NOT NULL, modified_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP on update CURRENT_TIMESTAMP NOT NULL, "
+ + "PRIMARY KEY (flow_group,flow_name,flow_execution_id,job_name,dag_action))";
// Deletes rows older than retention time period (in seconds) to prevent this table from growing unbounded.
private static final String RETENTION_STATEMENT = "DELETE FROM %s WHERE modified_time < DATE_SUB(CURRENT_TIMESTAMP, INTERVAL %s SECOND)";
@@ -97,13 +99,14 @@ public class MysqlDagActionStore implements DagActionStore {
}
@Override
- public boolean exists(String flowGroup, String flowName, String flowExecutionId, FlowActionType flowActionType) throws IOException, SQLException {
+ public boolean exists(String flowGroup, String flowName, String flowExecutionId, String jobName, DagActionType dagActionType) throws IOException, SQLException {
return dbStatementExecutor.withPreparedStatement(String.format(EXISTS_STATEMENT, tableName), existStatement -> {
int i = 0;
existStatement.setString(++i, flowGroup);
existStatement.setString(++i, flowName);
existStatement.setString(++i, flowExecutionId);
- existStatement.setString(++i, flowActionType.toString());
+ existStatement.setString(++i, jobName);
+ existStatement.setString(++i, dagActionType.toString());
ResultSet rs = null;
try {
rs = existStatement.executeQuery();
@@ -111,7 +114,7 @@ public class MysqlDagActionStore implements DagActionStore {
return rs.getBoolean(1);
} catch (SQLException e) {
throw new IOException(String.format("Failure checking existence of DagAction: %s in table %s",
- new DagAction(flowGroup, flowName, flowExecutionId, flowActionType), tableName), e);
+ new DagAction(flowGroup, flowName, flowExecutionId, jobName, dagActionType), tableName), e);
} finally {
if (rs != null) {
rs.close();
@@ -121,7 +124,12 @@ public class MysqlDagActionStore implements DagActionStore {
}
@Override
- public void addDagAction(String flowGroup, String flowName, String flowExecutionId, FlowActionType flowActionType)
+ public boolean exists(String flowGroup, String flowName, String flowExecutionId, DagActionType dagActionType) throws IOException, SQLException {
+ return exists(flowGroup, flowName, flowExecutionId, NO_JOB_NAME_DEFAULT, dagActionType);
+ }
+
+ @Override
+ public void addDagAction(String flowGroup, String flowName, String flowExecutionId, String jobName, DagActionType dagActionType)
throws IOException {
dbStatementExecutor.withPreparedStatement(String.format(INSERT_STATEMENT, tableName), insertStatement -> {
try {
@@ -129,14 +137,21 @@ public class MysqlDagActionStore implements DagActionStore {
insertStatement.setString(++i, flowGroup);
insertStatement.setString(++i, flowName);
insertStatement.setString(++i, flowExecutionId);
- insertStatement.setString(++i, flowActionType.toString());
+ insertStatement.setString(++i, jobName);
+ insertStatement.setString(++i, dagActionType.toString());
return insertStatement.executeUpdate();
} catch (SQLException e) {
throw new IOException(String.format("Failure adding action for DagAction: %s in table %s",
- new DagAction(flowGroup, flowName, flowExecutionId, flowActionType), tableName), e);
+ new DagAction(flowGroup, flowName, flowExecutionId, jobName, dagActionType), tableName), e);
}}, true);
}
+ @Override
+ public void addDagAction(String flowGroup, String flowName, String flowExecutionId, DagActionType dagActionType)
+ throws IOException {
+ addDagAction(flowGroup, flowName, flowExecutionId, NO_JOB_NAME_DEFAULT, dagActionType);
+ }
+
@Override
public boolean deleteDagAction(DagAction dagAction) throws IOException {
return dbStatementExecutor.withPreparedStatement(String.format(DELETE_STATEMENT, tableName), deleteStatement -> {
@@ -145,7 +160,8 @@ public class MysqlDagActionStore implements DagActionStore {
deleteStatement.setString(++i, dagAction.getFlowGroup());
deleteStatement.setString(++i, dagAction.getFlowName());
deleteStatement.setString(++i, dagAction.getFlowExecutionId());
- deleteStatement.setString(++i, dagAction.getFlowActionType().toString());
+ deleteStatement.setString(++i, dagAction.getJobName());
+ deleteStatement.setString(++i, dagAction.getDagActionType().toString());
int result = deleteStatement.executeUpdate();
return result != 0;
} catch (SQLException e) {
@@ -155,27 +171,27 @@ public class MysqlDagActionStore implements DagActionStore {
}
// TODO: later change this to getDagActions relating to a particular flow execution if it makes sense
- private DagAction getDagActionWithRetry(String flowGroup, String flowName, String flowExecutionId, FlowActionType flowActionType, ExponentialBackoff exponentialBackoff)
+ private DagAction getDagActionWithRetry(String flowGroup, String flowName, String flowExecutionId, String jobName, DagActionType dagActionType, ExponentialBackoff exponentialBackoff)
throws IOException, SQLException {
return dbStatementExecutor.withPreparedStatement(String.format(GET_STATEMENT, tableName), getStatement -> {
int i = 0;
getStatement.setString(++i, flowGroup);
getStatement.setString(++i, flowName);
getStatement.setString(++i, flowExecutionId);
- getStatement.setString(++i, flowActionType.toString());
+ getStatement.setString(++i, dagActionType.toString());
try (ResultSet rs = getStatement.executeQuery()) {
if (rs.next()) {
- return new DagAction(rs.getString(1), rs.getString(2), rs.getString(3), FlowActionType.valueOf(rs.getString(4)));
+ return new DagAction(rs.getString(1), rs.getString(2), rs.getString(3), rs.getString(4), DagActionType.valueOf(rs.getString(5)));
} else if (exponentialBackoff.awaitNextRetryIfAvailable()) {
- return getDagActionWithRetry(flowGroup, flowName, flowExecutionId, flowActionType, exponentialBackoff);
+ return getDagActionWithRetry(flowGroup, flowName, flowExecutionId, jobName, dagActionType, exponentialBackoff);
} else {
log.warn(String.format("Can not find dag action: %s with flowGroup: %s, flowName: %s, flowExecutionId: %s",
- flowActionType, flowGroup, flowName, flowExecutionId));
+ dagActionType, flowGroup, flowName, flowExecutionId));
return null;
}
} catch (SQLException | InterruptedException e) {
throw new IOException(String.format("Failure get %s from table %s",
- new DagAction(flowGroup, flowName, flowExecutionId, flowActionType), tableName), e);
+ new DagAction(flowGroup, flowName, flowExecutionId, jobName, dagActionType), tableName), e);
}
}, true);
}
@@ -186,7 +202,7 @@ public class MysqlDagActionStore implements DagActionStore {
HashSet<DagAction> result = new HashSet<>();
try (ResultSet rs = getAllStatement.executeQuery()) {
while (rs.next()) {
- result.add(new DagAction(rs.getString(1), rs.getString(2), rs.getString(3), FlowActionType.valueOf(rs.getString(4))));
+ result.add(new DagAction(rs.getString(1), rs.getString(2), rs.getString(3), rs.getString(4), DagActionType.valueOf(rs.getString(5))));
}
return result;
} catch (SQLException e) {
diff --git a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/util/InjectionNames.java b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/util/InjectionNames.java
index 55aa2130d..733f6ba07 100644
--- a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/util/InjectionNames.java
+++ b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/util/InjectionNames.java
@@ -33,4 +33,5 @@ public final class InjectionNames {
public static final String WARM_STANDBY_ENABLED = "statelessRestAPIEnabled";
public static final String MULTI_ACTIVE_SCHEDULER_ENABLED = "multiActiveSchedulerEnabled";
public static final String DAG_PROC_ENGINE_ENABLED = ServiceConfigKeys.DAG_PROCESSING_ENGINE_ENABLED;
+ public static final String MULTI_ACTIVE_EXECUTION_ENABLED = "multiActiveExecutionEnabled";
}
diff --git a/gobblin-runtime/src/main/java/org/apache/gobblin/scheduler/SchedulerService.java b/gobblin-runtime/src/main/java/org/apache/gobblin/scheduler/SchedulerService.java
index 285eb4449..c19cf22d0 100644
--- a/gobblin-runtime/src/main/java/org/apache/gobblin/scheduler/SchedulerService.java
+++ b/gobblin-runtime/src/main/java/org/apache/gobblin/scheduler/SchedulerService.java
@@ -21,6 +21,7 @@ import java.util.Properties;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
+import org.quartz.SchedulerFactory;
import org.quartz.impl.StdSchedulerFactory;
import com.google.common.base.Optional;
@@ -43,35 +44,52 @@ import org.apache.gobblin.util.PropertiesUtils;
@Singleton
public class SchedulerService extends AbstractIdleService {
+ private SchedulerFactory schedulerFactory;
+ // Refers to traditional job scheduler
@Getter
private Scheduler scheduler;
private final boolean waitForJobCompletion;
private final Optional<Properties> quartzProps;
- public SchedulerService(boolean waitForJobCompletion, Optional<Properties> quartzConfig) {
+ public SchedulerService(boolean waitForJobCompletion, Optional<Properties> quartzConfig,
+ StdSchedulerFactory schedulerFactory) {
this.waitForJobCompletion = waitForJobCompletion;
this.quartzProps = quartzConfig;
+ if (this.schedulerFactory == null) {
+ this.schedulerFactory = new StdSchedulerFactory();
+ }
+ else {
+ this.schedulerFactory = schedulerFactory;
+ }
}
- public SchedulerService(Properties props) {
+ public SchedulerService(Properties props, StdSchedulerFactory schedulerFactory) {
this(Boolean.parseBoolean(
props.getProperty(ConfigurationKeys.SCHEDULER_WAIT_FOR_JOB_COMPLETION_KEY,
ConfigurationKeys.DEFAULT_SCHEDULER_WAIT_FOR_JOB_COMPLETION)),
- Optional.of(PropertiesUtils.extractPropertiesWithPrefix(props, Optional.of("org.quartz."))));
+ Optional.of(PropertiesUtils.extractPropertiesWithPrefix(props, Optional.of("org.quartz."))), schedulerFactory);
+ }
+
+ public SchedulerService(Properties props) {
+ this(props, null);
}
- @Inject
public SchedulerService(Config cfg) {
+ this(cfg, null);
+ }
+
+ @Inject
+ public SchedulerService(Config cfg, StdSchedulerFactory schedulerFactory) {
this(cfg.hasPath(ConfigurationKeys.SCHEDULER_WAIT_FOR_JOB_COMPLETION_KEY) ?
- cfg.getBoolean(ConfigurationKeys.SCHEDULER_WAIT_FOR_JOB_COMPLETION_KEY) :
- Boolean.parseBoolean(ConfigurationKeys.DEFAULT_SCHEDULER_WAIT_FOR_JOB_COMPLETION),
- Optional.of(ConfigUtils.configToProperties(cfg, "org.quartz.")));
+ cfg.getBoolean(ConfigurationKeys.SCHEDULER_WAIT_FOR_JOB_COMPLETION_KEY) :
+ Boolean.parseBoolean(ConfigurationKeys.DEFAULT_SCHEDULER_WAIT_FOR_JOB_COMPLETION),
+ Optional.of(ConfigUtils.configToProperties(cfg, "org.quartz.")), schedulerFactory);
}
@Override protected void startUp() throws SchedulerException {
- StdSchedulerFactory schedulerFactory = new StdSchedulerFactory();
if (this.quartzProps.isPresent() && this.quartzProps.get().size() > 0) {
- schedulerFactory.initialize(this.quartzProps.get());
+ // Cast to StdSchedulerFactory to reference initialization method that generic interface does not provide
+ ((StdSchedulerFactory) schedulerFactory).initialize(this.quartzProps.get());
}
this.scheduler = schedulerFactory.getScheduler();
this.scheduler.start();
diff --git a/gobblin-runtime/src/test/java/org/apache/gobblin/runtime/api/MysqlMultiActiveLeaseArbiterTest.java b/gobblin-runtime/src/test/java/org/apache/gobblin/runtime/api/MysqlMultiActiveLeaseArbiterTest.java
index a77949541..9247cdbd0 100644
--- a/gobblin-runtime/src/test/java/org/apache/gobblin/runtime/api/MysqlMultiActiveLeaseArbiterTest.java
+++ b/gobblin-runtime/src/test/java/org/apache/gobblin/runtime/api/MysqlMultiActiveLeaseArbiterTest.java
@@ -35,23 +35,25 @@ import static org.apache.gobblin.runtime.api.MysqlMultiActiveLeaseArbiter.*;
@Slf4j
public class MysqlMultiActiveLeaseArbiterTest {
- private static final int EPSILON = 10000;
- private static final int MORE_THAN_EPSILON = (int) (EPSILON * 1.1);
- private static final int LINGER = 50000;
+ private static final long EPSILON = 10000;
+ private static final long MORE_THAN_EPSILON = (long) (EPSILON * 1.1);
+ private static final long LINGER = 50000;
+ private static final long MORE_THAN_LINGER = (long) (LINGER * 1.1);
private static final String USER = "testUser";
private static final String PASSWORD = "testPassword";
private static final String TABLE = "mysql_multi_active_lease_arbiter_store";
private static final String flowGroup = "testFlowGroup";
private static final String flowGroup2 = "testFlowGroup2";
private static final String flowName = "testFlowName";
+ private static final String jobName = "testJobName";
private static final String flowExecutionId = "12345677";
// Dag actions with the same flow info but different flow action types are considered unique
private static DagActionStore.DagAction launchDagAction =
- new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId, DagActionStore.FlowActionType.LAUNCH);
+ new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId, jobName, DagActionStore.DagActionType.LAUNCH);
private static DagActionStore.DagAction resumeDagAction =
- new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId, DagActionStore.FlowActionType.RESUME);
+ new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId, jobName, DagActionStore.DagActionType.RESUME);
private static DagActionStore.DagAction launchDagAction2 =
- new DagActionStore.DagAction(flowGroup2, flowName, flowExecutionId, DagActionStore.FlowActionType.LAUNCH);
+ new DagActionStore.DagAction(flowGroup2, flowName, flowExecutionId, jobName, DagActionStore.DagActionType.LAUNCH);
private static final long eventTimeMillis = System.currentTimeMillis();
private static final Timestamp dummyTimestamp = new Timestamp(99999);
private MysqlMultiActiveLeaseArbiter mysqlMultiActiveLeaseArbiter;
@@ -94,13 +96,13 @@ public class MysqlMultiActiveLeaseArbiterTest {
(MultiActiveLeaseArbiter.LeaseObtainedStatus) firstLaunchStatus;
Assert.assertTrue(firstObtainedStatus.getEventTimeMillis() <=
firstObtainedStatus.getLeaseAcquisitionTimestamp());
- Assert.assertTrue(firstObtainedStatus.getFlowAction().equals(
+ Assert.assertTrue(firstObtainedStatus.getDagAction().equals(
new DagActionStore.DagAction(flowGroup, flowName, String.valueOf(firstObtainedStatus.getEventTimeMillis()),
- DagActionStore.FlowActionType.LAUNCH)));
+ jobName, DagActionStore.DagActionType.LAUNCH)));
// Verify that different DagAction types for the same flow can have leases at the same time
DagActionStore.DagAction killDagAction = new
- DagActionStore.DagAction(flowGroup, flowName, flowExecutionId, DagActionStore.FlowActionType.KILL);
+ DagActionStore.DagAction(flowGroup, flowName, flowExecutionId, jobName, DagActionStore.DagActionType.KILL);
MultiActiveLeaseArbiter.LeaseAttemptStatus killStatus =
mysqlMultiActiveLeaseArbiter.tryAcquireLease(killDagAction, eventTimeMillis, false, true);
Assert.assertTrue(killStatus instanceof MultiActiveLeaseArbiter.LeaseObtainedStatus);
@@ -224,7 +226,7 @@ public class MysqlMultiActiveLeaseArbiterTest {
DagActionStore.DagAction updatedResumeDagAction = resumeDagAction.updateFlowExecutionId(
selectInfoResult.getEventTimeMillis());
boolean markedSuccess = mysqlMultiActiveLeaseArbiter.recordLeaseSuccess(new LeaseObtainedStatus(
- updatedResumeDagAction, selectInfoResult.getLeaseAcquisitionTimeMillis().get()));
+ updatedResumeDagAction, selectInfoResult.getLeaseAcquisitionTimeMillis().get(), LINGER, null));
Assert.assertTrue(markedSuccess);
// Ensure no NPE results from calling this after a lease has been completed and acquisition timestamp val is NULL
mysqlMultiActiveLeaseArbiter.evaluateStatusAfterLeaseAttempt(1, resumeDagAction,
@@ -283,7 +285,7 @@ public class MysqlMultiActiveLeaseArbiterTest {
// Read database to obtain existing db eventTimeMillis and wait enough time for the lease to expire
MysqlMultiActiveLeaseArbiter.SelectInfoResult selectInfoResult =
mysqlMultiActiveLeaseArbiter.getRowInfo(resumeDagAction);
- Thread.sleep(LINGER);
+ Thread.sleep(MORE_THAN_LINGER);
LeaseAttemptStatus attemptStatus =
mysqlMultiActiveLeaseArbiter.tryAcquireLease(resumeDagAction, selectInfoResult.getEventTimeMillis(), true, true);
Assert.assertTrue(attemptStatus instanceof LeaseObtainedStatus);
@@ -306,7 +308,7 @@ public class MysqlMultiActiveLeaseArbiterTest {
DagActionStore.DagAction updatedResumeDagAction = resumeDagAction.updateFlowExecutionId(
selectInfoResult.getEventTimeMillis());
boolean markedSuccess = mysqlMultiActiveLeaseArbiter.recordLeaseSuccess(new LeaseObtainedStatus(
- updatedResumeDagAction, selectInfoResult.getLeaseAcquisitionTimeMillis().get()));
+ updatedResumeDagAction, selectInfoResult.getLeaseAcquisitionTimeMillis().get(), LINGER, null));
Assert.assertTrue(markedSuccess);
// Sleep enough time for the event to have been considered distinct
@@ -331,8 +333,8 @@ public class MysqlMultiActiveLeaseArbiterTest {
MultiActiveLeaseArbiter.LeaseObtainedStatus firstObtainedStatus =
(MultiActiveLeaseArbiter.LeaseObtainedStatus) firstLaunchStatus;
Assert.assertTrue(firstObtainedStatus.getEventTimeMillis() <= firstObtainedStatus.getLeaseAcquisitionTimestamp());
- Assert.assertTrue(firstObtainedStatus.getFlowAction().equals(
- new DagActionStore.DagAction(flowGroup2, flowName, flowExecutionId, DagActionStore.FlowActionType.LAUNCH)));
+ Assert.assertTrue(firstObtainedStatus.getDagAction().equals(
+ new DagActionStore.DagAction(flowGroup2, flowName, flowExecutionId, jobName, DagActionStore.DagActionType.LAUNCH)));
// A second attempt to obtain a lease on the same action should return a LeasedToAnotherStatus which also contains
// the original flowExecutionId
@@ -342,7 +344,7 @@ public class MysqlMultiActiveLeaseArbiterTest {
MultiActiveLeaseArbiter.LeasedToAnotherStatus secondLeasedToAnotherStatus =
(MultiActiveLeaseArbiter.LeasedToAnotherStatus) secondLaunchStatus;
Assert.assertEquals(firstObtainedStatus.getEventTimeMillis(), secondLeasedToAnotherStatus.getEventTimeMillis());
- Assert.assertTrue(firstObtainedStatus.getFlowAction().equals(
- new DagActionStore.DagAction(flowGroup2, flowName, flowExecutionId, DagActionStore.FlowActionType.LAUNCH)));
+ Assert.assertTrue(firstObtainedStatus.getDagAction().equals(
+ new DagActionStore.DagAction(flowGroup2, flowName, flowExecutionId, jobName, DagActionStore.DagActionType.LAUNCH)));
}
}
diff --git a/gobblin-runtime/src/test/java/org/apache/gobblin/runtime/dag_action_store/MysqlDagActionStoreTest.java b/gobblin-runtime/src/test/java/org/apache/gobblin/runtime/dag_action_store/MysqlDagActionStoreTest.java
index 255dd0789..944cfa027 100644
--- a/gobblin-runtime/src/test/java/org/apache/gobblin/runtime/dag_action_store/MysqlDagActionStoreTest.java
+++ b/gobblin-runtime/src/test/java/org/apache/gobblin/runtime/dag_action_store/MysqlDagActionStoreTest.java
@@ -40,6 +40,8 @@ public class MysqlDagActionStoreTest {
private static final String TABLE = "dag_action_store";
private static final String flowGroup = "testFlowGroup";
private static final String flowName = "testFlowName";
+ private static final String jobName = "testJobName";
+ private static final String jobName_2 = "testJobName2";
private static final String flowExecutionId = "12345677";
private static final String flowExecutionId_2 = "12345678";
private static final String flowExecutionId_3 = "12345679";
@@ -61,22 +63,23 @@ public class MysqlDagActionStoreTest {
@Test
public void testAddAction() throws Exception {
- this.mysqlDagActionStore.addDagAction(flowGroup, flowName, flowExecutionId, DagActionStore.FlowActionType.KILL);
+ this.mysqlDagActionStore.addDagAction(flowGroup, flowName, flowExecutionId, jobName, DagActionStore.DagActionType.KILL);
//Should not be able to add KILL again when previous one exist
Assert.expectThrows(IOException.class,
- () -> this.mysqlDagActionStore.addDagAction(flowGroup, flowName, flowExecutionId, DagActionStore.FlowActionType.KILL));
+ () -> this.mysqlDagActionStore.addDagAction(flowGroup, flowName, flowExecutionId, jobName, DagActionStore.DagActionType.KILL));
//Should be able to add a RESUME action for same execution as well as KILL for another execution of the flow
- this.mysqlDagActionStore.addDagAction(flowGroup, flowName, flowExecutionId, DagActionStore.FlowActionType.RESUME);
- this.mysqlDagActionStore.addDagAction(flowGroup, flowName, flowExecutionId_2, DagActionStore.FlowActionType.KILL);
+ this.mysqlDagActionStore.addDagAction(flowGroup, flowName, flowExecutionId, jobName, DagActionStore.DagActionType.RESUME);
+ this.mysqlDagActionStore.addDagAction(flowGroup, flowName, flowExecutionId_2, jobName, DagActionStore.DagActionType.KILL);
}
@Test(dependsOnMethods = "testAddAction")
public void testExists() throws Exception {
- Assert.assertTrue(this.mysqlDagActionStore.exists(flowGroup, flowName, flowExecutionId, DagActionStore.FlowActionType.KILL));
- Assert.assertTrue(this.mysqlDagActionStore.exists(flowGroup, flowName, flowExecutionId, DagActionStore.FlowActionType.RESUME));
- Assert.assertTrue(this.mysqlDagActionStore.exists(flowGroup, flowName, flowExecutionId_2, DagActionStore.FlowActionType.KILL));
- Assert.assertFalse(this.mysqlDagActionStore.exists(flowGroup, flowName, flowExecutionId_3, DagActionStore.FlowActionType.RESUME));
- Assert.assertFalse(this.mysqlDagActionStore.exists(flowGroup, flowName, flowExecutionId_3, DagActionStore.FlowActionType.KILL));
+ Assert.assertTrue(this.mysqlDagActionStore.exists(flowGroup, flowName, flowExecutionId, jobName, DagActionStore.DagActionType.KILL));
+ Assert.assertTrue(this.mysqlDagActionStore.exists(flowGroup, flowName, flowExecutionId, jobName, DagActionStore.DagActionType.RESUME));
+ Assert.assertTrue(this.mysqlDagActionStore.exists(flowGroup, flowName, flowExecutionId_2, jobName, DagActionStore.DagActionType.KILL));
+ Assert.assertFalse(this.mysqlDagActionStore.exists(flowGroup, flowName, flowExecutionId_2, jobName_2, DagActionStore.DagActionType.KILL));
+ Assert.assertFalse(this.mysqlDagActionStore.exists(flowGroup, flowName, flowExecutionId_3, jobName, DagActionStore.DagActionType.RESUME));
+ Assert.assertFalse(this.mysqlDagActionStore.exists(flowGroup, flowName, flowExecutionId_3, jobName, DagActionStore.DagActionType.KILL));
}
@Test(dependsOnMethods = "testExists")
@@ -84,20 +87,20 @@ public class MysqlDagActionStoreTest {
Collection<DagActionStore.DagAction> dagActions = this.mysqlDagActionStore.getDagActions();
Assert.assertEquals(3, dagActions.size());
HashSet<DagActionStore.DagAction> set = new HashSet<>();
- set.add(new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId, DagActionStore.FlowActionType.KILL));
- set.add(new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId, DagActionStore.FlowActionType.RESUME));
- set.add(new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId_2, DagActionStore.FlowActionType.KILL));
+ set.add(new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId, jobName, DagActionStore.DagActionType.KILL));
+ set.add(new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId, jobName, DagActionStore.DagActionType.RESUME));
+ set.add(new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId_2, jobName, DagActionStore.DagActionType.KILL));
Assert.assertEquals(dagActions, set);
}
@Test(dependsOnMethods = "testGetActions")
public void testDeleteAction() throws IOException, SQLException {
this.mysqlDagActionStore.deleteDagAction(
- new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId, DagActionStore.FlowActionType.KILL));
+ new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId, jobName, DagActionStore.DagActionType.KILL));
Assert.assertEquals(this.mysqlDagActionStore.getDagActions().size(), 2);
- Assert.assertFalse(this.mysqlDagActionStore.exists(flowGroup, flowName, flowExecutionId, DagActionStore.FlowActionType.KILL));
- Assert.assertTrue(this.mysqlDagActionStore.exists(flowGroup, flowName, flowExecutionId, DagActionStore.FlowActionType.RESUME));
- Assert.assertTrue(this.mysqlDagActionStore.exists(flowGroup, flowName, flowExecutionId_2, DagActionStore.FlowActionType.KILL));
+ Assert.assertFalse(this.mysqlDagActionStore.exists(flowGroup, flowName, flowExecutionId, jobName, DagActionStore.DagActionType.KILL));
+ Assert.assertTrue(this.mysqlDagActionStore.exists(flowGroup, flowName, flowExecutionId, jobName, DagActionStore.DagActionType.RESUME));
+ Assert.assertTrue(this.mysqlDagActionStore.exists(flowGroup, flowName, flowExecutionId_2, jobName, DagActionStore.DagActionType.KILL));
}
}
\ No newline at end of file
diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/core/GobblinServiceConfiguration.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/core/GobblinServiceConfiguration.java
index 83b64c900..54f8779b2 100644
--- a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/core/GobblinServiceConfiguration.java
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/core/GobblinServiceConfiguration.java
@@ -46,6 +46,9 @@ public class GobblinServiceConfiguration {
@Getter
private final boolean isMultiActiveSchedulerEnabled;
+ @Getter
+ private final boolean isMultiActiveExecutionEnabled;
+
@Getter
private final boolean isFlowCatalogEnabled;
@@ -106,6 +109,7 @@ public class GobblinServiceConfiguration {
this.isWarmStandbyEnabled = ConfigUtils.getBoolean(config, ServiceConfigKeys.GOBBLIN_SERVICE_WARM_STANDBY_ENABLED_KEY, false);
this.isMultiActiveSchedulerEnabled = ConfigUtils.getBoolean(config, ServiceConfigKeys.GOBBLIN_SERVICE_MULTI_ACTIVE_SCHEDULER_ENABLED_KEY, false);
+ this.isMultiActiveExecutionEnabled = ConfigUtils.getBoolean(config, ServiceConfigKeys.GOBBLIN_SERVICE_MULTI_ACTIVE_EXECUTION_ENABLED, false);
this.isHelixManagerEnabled = config.hasPath(ServiceConfigKeys.ZK_CONNECTION_STRING_KEY);
this.isJobStatusMonitorEnabled =
diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/core/GobblinServiceGuiceModule.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/core/GobblinServiceGuiceModule.java
index 01b2598b0..414055faf 100644
--- a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/core/GobblinServiceGuiceModule.java
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/core/GobblinServiceGuiceModule.java
@@ -20,6 +20,7 @@ package org.apache.gobblin.service.modules.core;
import java.util.Objects;
import org.apache.helix.HelixManager;
+import org.quartz.impl.StdSchedulerFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -74,8 +75,10 @@ import org.apache.gobblin.service.modules.orchestration.DagManager;
import org.apache.gobblin.service.modules.orchestration.DagProcFactory;
import org.apache.gobblin.service.modules.orchestration.DagProcessingEngine;
import org.apache.gobblin.service.modules.orchestration.DagTaskStream;
-import org.apache.gobblin.service.modules.orchestration.FlowTriggerHandler;
import org.apache.gobblin.service.modules.orchestration.MostlyMySqlDagManagementStateStore;
+import org.apache.gobblin.service.modules.orchestration.DagActionReminderScheduler;
+import org.apache.gobblin.service.modules.orchestration.ReminderSettingDagProcLeaseArbiter;
+import org.apache.gobblin.service.modules.orchestration.FlowLaunchHandler;
import org.apache.gobblin.service.modules.orchestration.Orchestrator;
import org.apache.gobblin.service.modules.orchestration.UserQuotaManager;
import org.apache.gobblin.service.modules.restli.GobblinServiceFlowConfigResourceHandler;
@@ -105,6 +108,8 @@ import org.apache.gobblin.util.ConfigUtils;
public class GobblinServiceGuiceModule implements Module {
+ public static final String SCHEDULER_LEASE_ARBITER_NAME = "SchedulerLeaseArbiter";
+ public static final String EXECUTOR_LEASE_ARBITER_NAME = "ExecutorLeaseArbiter";
private static final Logger LOGGER = LoggerFactory.getLogger(GobblinServiceGuiceModule.class);
private static final String JOB_STATUS_RETRIEVER_CLASS_KEY = "jobStatusRetriever.class";
@@ -166,6 +171,9 @@ public class GobblinServiceGuiceModule implements Module {
binder.bindConstant()
.annotatedWith(Names.named(InjectionNames.DAG_PROC_ENGINE_ENABLED))
.to(serviceConfig.isDagProcessingEngineEnabled());
+ binder.bindConstant()
+ .annotatedWith(Names.named(InjectionNames.MULTI_ACTIVE_EXECUTION_ENABLED))
+ .to(serviceConfig.isMultiActiveExecutionEnabled());
OptionalBinder.newOptionalBinder(binder, DagActionStore.class);
if (serviceConfig.isWarmStandbyEnabled()) {
@@ -180,17 +188,36 @@ public class GobblinServiceGuiceModule implements Module {
}
OptionalBinder.newOptionalBinder(binder, MultiActiveLeaseArbiter.class);
- OptionalBinder.newOptionalBinder(binder, FlowTriggerHandler.class);
+ OptionalBinder.newOptionalBinder(binder, FlowLaunchHandler.class);
if (serviceConfig.isMultiActiveSchedulerEnabled()) {
- binder.bind(MultiActiveLeaseArbiter.class).to(MysqlMultiActiveLeaseArbiter.class);
- binder.bind(FlowTriggerHandler.class);
+ binder.bind(MultiActiveLeaseArbiter.class).annotatedWith(Names.named(SCHEDULER_LEASE_ARBITER_NAME)).to(MysqlMultiActiveLeaseArbiter.class);
+ binder.bind(FlowLaunchHandler.class);
}
- binder.bind(DagManagement.class).to(DagManagementTaskStreamImpl.class);
- binder.bind(DagTaskStream.class).to(DagManagementTaskStreamImpl.class);
- binder.bind(DagManagementStateStore.class).to(MostlyMySqlDagManagementStateStore.class).in(Singleton.class);
- binder.bind(DagProcFactory.class);
- binder.bind(DagProcessingEngine.class);
+ // Note: only one SchedulerFactory instance should exist per JVM
+ binder.bind(StdSchedulerFactory.class).in(Singleton.class);
+
+ OptionalBinder.newOptionalBinder(binder, DagManagement.class);
+ OptionalBinder.newOptionalBinder(binder, DagTaskStream.class);
+ OptionalBinder.newOptionalBinder(binder, DagManagementStateStore.class);
+ OptionalBinder.newOptionalBinder(binder, DagProcFactory.class);
+ OptionalBinder.newOptionalBinder(binder, DagProcessingEngine.class);
+ if (serviceConfig.isDagProcessingEngineEnabled()) {
+ binder.bind(DagManagement.class).to(DagManagementTaskStreamImpl.class);
+ binder.bind(DagTaskStream.class).to(DagManagementTaskStreamImpl.class);
+ binder.bind(DagManagementStateStore.class).to(MostlyMySqlDagManagementStateStore.class).in(Singleton.class);
+ binder.bind(DagProcFactory.class);
+ binder.bind(DagProcessingEngine.class);
+
+ // Multi-active execution is only compatible with dagProcessingEngine configuration
+ OptionalBinder.newOptionalBinder(binder, ReminderSettingDagProcLeaseArbiter.class);
+ OptionalBinder.newOptionalBinder(binder, DagActionReminderScheduler.class);
+ if (serviceConfig.isMultiActiveExecutionEnabled()) {
+ binder.bind(MultiActiveLeaseArbiter.class).annotatedWith(Names.named(EXECUTOR_LEASE_ARBITER_NAME)).to(MysqlMultiActiveLeaseArbiter.class);
+ binder.bind(DagActionReminderScheduler.class);
+ binder.bind(ReminderSettingDagProcLeaseArbiter.class);
+ }
+ }
binder.bind(FlowConfigsResource.class);
binder.bind(FlowConfigsV2Resource.class);
diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagActionReminderScheduler.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagActionReminderScheduler.java
new file mode 100644
index 000000000..f1a3109ee
--- /dev/null
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagActionReminderScheduler.java
@@ -0,0 +1,76 @@
+/*
+ * 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.util.Optional;
+
+import org.quartz.JobDetail;
+import org.quartz.Scheduler;
+import org.quartz.SchedulerException;
+import org.quartz.Trigger;
+import org.quartz.impl.StdSchedulerFactory;
+
+import javax.inject.Inject;
+
+import org.apache.gobblin.runtime.api.DagActionStore;
+
+
+/**
+ * This class is used to keep track of reminders of pending flow action events to execute. A host calls the
+ * {#scheduleReminderJob} on a flow action that it failed to acquire a lease on but has not yet completed. The reminder
+ * will fire once the previous lease owner's lease is expected to expire.
+ */
+public class DagActionReminderScheduler {
+ public static final String DAG_ACTION_REMINDER_SCHEDULER_KEY = "DagActionReminderScheduler";
+ private final Scheduler quartzScheduler;
+ private final Optional<DagManagement> dagManagement;
+
+ @Inject
+ public DagActionReminderScheduler(StdSchedulerFactory schedulerFactory, Optional<DagManagement> dagManagement)
+ throws SchedulerException {
+ // Create a new Scheduler to be used solely for the DagProc reminders
+ this.quartzScheduler = schedulerFactory.getScheduler(DAG_ACTION_REMINDER_SCHEDULER_KEY);
+ this.dagManagement = dagManagement;
+ }
+
+ /**
+ * Uses a dagAction & reminder duration in milliseconds to create a reminder job that will fire
+ * `reminderDurationMillis` after the current time
+ * @param dagAction
+ * @param reminderDurationMillis
+ * @throws SchedulerException
+ */
+ public void scheduleReminder(DagActionStore.DagAction dagAction, long reminderDurationMillis)
+ throws SchedulerException {
+ if (!dagManagement.isPresent()) {
+ throw new RuntimeException("DagManagement not initialized in multi-active execution mode when required.");
+ }
+ JobDetail jobDetail = ReminderSettingDagProcLeaseArbiter.createReminderJobDetail(dagManagement.get(), dagAction);
+ Trigger trigger = ReminderSettingDagProcLeaseArbiter.createReminderJobTrigger(dagAction, reminderDurationMillis);
+ quartzScheduler.scheduleJob(jobDetail, trigger);
+ }
+
+ public void unscheduleReminderJob(DagActionStore.DagAction dagAction) throws SchedulerException {
+ if (!dagManagement.isPresent()) {
+ throw new RuntimeException("DagManagement not initialized in multi-active execution mode when required.");
+ }
+ JobDetail jobDetail = ReminderSettingDagProcLeaseArbiter.createReminderJobDetail(dagManagement.get(), dagAction);
+ quartzScheduler.deleteJob(jobDetail.getKey());
+ }
+
+}
diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementTaskStreamImpl.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementTaskStreamImpl.java
index 63a8fba0e..192243075 100644
--- a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementTaskStreamImpl.java
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementTaskStreamImpl.java
@@ -53,14 +53,18 @@ public class DagManagementTaskStreamImpl implements DagManagement, DagTaskStream
@Inject(optional=true)
protected Optional<DagActionStore> dagActionStore;
+ protected Optional<ReminderSettingDagProcLeaseArbiter> reminderSettingDagProcLeaseArbiter;
@Inject
private static final int MAX_HOUSEKEEPING_THREAD_DELAY = 180;
private final BlockingQueue<DagActionStore.DagAction> dagActionQueue = new LinkedBlockingQueue<>();
+ // TODO: need to pass reference to DagProcLeaseArbiter without creating a circular reference in Guice
@Inject
- public DagManagementTaskStreamImpl(Config config, Optional<DagActionStore> dagActionStore) {
+ public DagManagementTaskStreamImpl(Config config, Optional<DagActionStore> dagActionStore,
+ Optional<ReminderSettingDagProcLeaseArbiter> reminderSettingDagProcLeaseArbiter) {
this.config = config;
this.dagActionStore = dagActionStore;
+ this.reminderSettingDagProcLeaseArbiter = reminderSettingDagProcLeaseArbiter;
MetricContext metricContext = Instrumented.getMetricContext(ConfigUtils.configToState(ConfigFactory.empty()), getClass());
this.eventSubmitter = new EventSubmitter.Builder(metricContext, "org.apache.gobblin.service").build();
}
@@ -82,17 +86,21 @@ public class DagManagementTaskStreamImpl implements DagManagement, DagTaskStream
@Override
public DagTask next() {
+ /*TODO: this requires use of lease arbiter, in single-active execution lease arbiter will not be present and we can
+ provide a dummy LeaseObtainedStatus or create alternate route
+ */
+ if (!this.reminderSettingDagProcLeaseArbiter.isPresent()) {
+ throw new RuntimeException("DagManagement not initialized in multi-active execution mode when required.");
+ }
try {
- DagActionStore.DagAction dagAction = this.dagActionQueue.take(); //`take` blocks till element is not available
- // todo reconsider the use of MultiActiveLeaseArbiter
- //MultiActiveLeaseArbiter.LeaseAttemptStatus leaseAttemptStatus = new MultiActiveLeaseArbiter.LeaseObtainedStatus(dagAction);
- // todo - uncomment after flow trigger handler provides such an api
- //Properties jobProps = getJobProperties(dagAction);
- //flowTriggerHandler.getLeaseOnDagAction(jobProps, dagAction, System.currentTimeMillis());
- //if (leaseAttemptStatus instanceof MultiActiveLeaseArbiter.LeaseObtainedStatus) {
- // can it return null? is this iterator allowed to return null?
- return createDagTask(dagAction, new MultiActiveLeaseArbiter.LeaseObtainedStatus(dagAction, System.currentTimeMillis()));
- //}
+ MultiActiveLeaseArbiter.LeaseAttemptStatus leaseAttemptStatus = null;
+ DagActionStore.DagAction dagAction = null;
+ while (!(leaseAttemptStatus instanceof MultiActiveLeaseArbiter.LeaseObtainedStatus)) {
+ dagAction = this.dagActionQueue.take(); //`take` blocks till element is not available
+ // TODO: need to handle reminder events and flag them
+ leaseAttemptStatus = this.reminderSettingDagProcLeaseArbiter.get().tryAcquireLease(dagAction, System.currentTimeMillis(), false, false);
+ }
+ return createDagTask(dagAction, (MultiActiveLeaseArbiter.LeaseObtainedStatus) leaseAttemptStatus);
} catch (Throwable t) {
//TODO: need to handle exceptions gracefully
log.error("Error getting DagAction from the queue / creating DagTask", t);
@@ -101,9 +109,9 @@ public class DagManagementTaskStreamImpl implements DagManagement, DagTaskStream
}
private DagTask createDagTask(DagActionStore.DagAction dagAction, MultiActiveLeaseArbiter.LeaseObtainedStatus leaseObtainedStatus) {
- DagActionStore.FlowActionType flowActionType = dagAction.getFlowActionType();
+ DagActionStore.DagActionType dagActionType = dagAction.getDagActionType();
- switch (flowActionType) {
+ switch (dagActionType) {
case LAUNCH:
return new LaunchDagTask(dagAction, leaseObtainedStatus);
default:
diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java
index 78adc2694..71b8a318a 100644
--- a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java
@@ -183,8 +183,9 @@ public class DagManager extends AbstractIdleService {
return Joiner.on("_").join(flowGroup, flowName, flowExecutionId);
}
- DagActionStore.DagAction toDagAction(DagActionStore.FlowActionType actionType) {
- return new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId, actionType);
+ DagActionStore.DagAction toDagAction(DagActionStore.DagActionType actionType) {
+ // defaults to empty jobName
+ return new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId, "", actionType);
}
}
@@ -313,7 +314,7 @@ public class DagManager extends AbstractIdleService {
// After persisting the dag, its status will be tracked by active dagManagers so the action should be deleted
// to avoid duplicate executions upon leadership change
if (this.dagActionStore.isPresent()) {
- this.dagActionStore.get().deleteDagAction(dagId.toDagAction(DagActionStore.FlowActionType.LAUNCH));
+ this.dagActionStore.get().deleteDagAction(dagId.toDagAction(DagActionStore.DagActionType.LAUNCH));
}
}
int queueId = DagManagerUtils.getDagQueueId(dag, this.numThreads);
@@ -588,10 +589,10 @@ public class DagManager extends AbstractIdleService {
}
}
- private void removeDagActionFromStore(DagId dagId, DagActionStore.FlowActionType flowActionType) throws IOException {
+ private void removeDagActionFromStore(DagId dagId, DagActionStore.DagActionType dagActionType) throws IOException {
if (this.dagActionStore.isPresent()) {
this.dagActionStore.get().deleteDagAction(
- new DagActionStore.DagAction(dagId.flowGroup, dagId.flowName, dagId.flowExecutionId, flowActionType));
+ new DagActionStore.DagAction(dagId.flowGroup, dagId.flowName, dagId.flowExecutionId, "", dagActionType));
}
}
@@ -603,13 +604,13 @@ public class DagManager extends AbstractIdleService {
String dagId= dagIdToResume.toString();
if (!this.failedDagIds.contains(dagId)) {
log.warn("No dag found with dagId " + dagId + ", so cannot resume flow");
- removeDagActionFromStore(dagIdToResume, DagActionStore.FlowActionType.RESUME);
+ removeDagActionFromStore(dagIdToResume, DagActionStore.DagActionType.RESUME);
return;
}
Dag<JobExecutionPlan> dag = this.failedDagStateStore.getDag(dagId);
if (dag == null) {
log.error("Dag " + dagId + " was found in memory but not found in failed dag state store");
- removeDagActionFromStore(dagIdToResume, DagActionStore.FlowActionType.RESUME);
+ removeDagActionFromStore(dagIdToResume, DagActionStore.DagActionType.RESUME);
return;
}
@@ -663,7 +664,7 @@ public class DagManager extends AbstractIdleService {
if (dagReady) {
this.dagStateStore.writeCheckpoint(dag.getValue());
this.failedDagStateStore.cleanUp(dag.getValue());
- removeDagActionFromStore(DagManagerUtils.generateDagId(dag.getValue()), DagActionStore.FlowActionType.RESUME);
+ removeDagActionFromStore(DagManagerUtils.generateDagId(dag.getValue()), DagActionStore.DagActionType.RESUME);
this.failedDagIds.remove(dag.getKey());
this.resumingDags.remove(dag.getKey());
initialize(dag.getValue());
@@ -693,7 +694,7 @@ public class DagManager extends AbstractIdleService {
log.warn("Did not find Dag with id {}, it might be already cancelled/finished.", dagToCancel);
}
// Called after a KILL request is received
- removeDagActionFromStore(dagId, DagActionStore.FlowActionType.KILL);
+ removeDagActionFromStore(dagId, DagActionStore.DagActionType.KILL);
}
private void cancelDagNode(DagNode<JobExecutionPlan> dagNodeToCancel) throws ExecutionException, InterruptedException {
diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagerUtils.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagerUtils.java
index d9d928e19..decf579aa 100644
--- a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagerUtils.java
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagerUtils.java
@@ -68,17 +68,18 @@ public class DagManagerUtils {
return getFlowId(dag.getStartNodes().get(0));
}
- public static DagActionStore.DagAction createDagActionFromDag(Dag<JobExecutionPlan> dag, DagActionStore.FlowActionType flowActionType) {
- return createDagActionFromDagNode(dag.getStartNodes().get(0), flowActionType);
+ public static DagActionStore.DagAction createDagActionFromDag(Dag<JobExecutionPlan> dag, DagActionStore.DagActionType dagActionType) {
+ return createDagActionFromDagNode(dag.getStartNodes().get(0), dagActionType);
}
- // todo - dag action object does not have any identifier to tell if it is for a complete dag or just for one dag node
- public static DagActionStore.DagAction createDagActionFromDagNode(DagNode<JobExecutionPlan> dagNode, DagActionStore.FlowActionType flowActionType) {
+ // todo - verify if jobName will always be present or need default and update FlowId, DagId etc to contain jobName
+ public static DagActionStore.DagAction createDagActionFromDagNode(DagNode<JobExecutionPlan> dagNode, DagActionStore.DagActionType dagActionType) {
Config jobConfig = dagNode.getValue().getJobSpec().getConfig();
String flowGroup = jobConfig.getString(ConfigurationKeys.FLOW_GROUP_KEY);
String flowName = jobConfig.getString(ConfigurationKeys.FLOW_NAME_KEY);
String flowExecutionId = jobConfig.getString(ConfigurationKeys.FLOW_EXECUTION_ID_KEY);
- return new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId, flowActionType);
+ String jobName = jobConfig.getString(ConfigurationKeys.JOB_NAME_KEY);
+ return new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId, jobName, dagActionType);
}
static FlowId getFlowId(DagNode<JobExecutionPlan> dagNode) {
diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngine.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngine.java
index 989102898..aa4a5f0e0 100644
--- a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngine.java
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngine.java
@@ -17,11 +17,11 @@
package org.apache.gobblin.service.modules.orchestration;
+import java.util.Optional;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Optional;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import com.typesafe.config.Config;
@@ -52,23 +52,25 @@ import org.apache.gobblin.util.ExecutorsUtils;
@Singleton
public class DagProcessingEngine {
- @Getter private final DagTaskStream dagTaskStream;
- @Getter DagManagementStateStore dagManagementStateStore;
+ @Getter private final Optional<DagTaskStream> dagTaskStream;
+ @Getter Optional<DagManagementStateStore> dagManagementStateStore;
@Inject
- public DagProcessingEngine(Config config, DagTaskStream dagTaskStream, DagProcFactory dagProcFactory,
- DagManagementStateStore dagManagementStateStore) {
+ public DagProcessingEngine(Config config, Optional<DagTaskStream> dagTaskStream,
+ Optional<DagProcFactory> dagProcFactory, Optional<DagManagementStateStore> dagManagementStateStore) {
Integer numThreads = ConfigUtils.getInt
(config, ServiceConfigKeys.NUM_DAG_PROC_THREADS_KEY, ServiceConfigKeys.DEFAULT_NUM_DAG_PROC_THREADS);
ScheduledExecutorService scheduledExecutorPool =
Executors.newScheduledThreadPool(numThreads,
- ExecutorsUtils.newThreadFactory(Optional.of(log), Optional.of("DagProcessingEngineThread")));
+ ExecutorsUtils.newThreadFactory(com.google.common.base.Optional.of(log),
+ com.google.common.base.Optional.of("DagProcessingEngineThread")));
this.dagTaskStream = dagTaskStream;
this.dagManagementStateStore = dagManagementStateStore;
for (int i=0; i < numThreads; i++) {
// todo - set metrics for count of active DagProcEngineThread
- DagProcEngineThread dagProcEngineThread = new DagProcEngineThread(dagTaskStream, dagProcFactory, dagManagementStateStore);
+ DagProcEngineThread dagProcEngineThread = new DagProcEngineThread(dagTaskStream.get(), dagProcFactory.get(),
+ dagManagementStateStore.get());
scheduledExecutorPool.submit(dagProcEngineThread);
}
}
diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/FlowTriggerHandler.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/FlowLaunchHandler.java
similarity index 64%
rename from gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/FlowTriggerHandler.java
rename to gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/FlowLaunchHandler.java
index 04ca6f597..1cd5af044 100644
--- a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/FlowTriggerHandler.java
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/FlowLaunchHandler.java
@@ -30,6 +30,7 @@ import java.util.Locale;
import java.util.Properties;
import java.util.Random;
import javax.inject.Inject;
+import javax.inject.Named;
import lombok.extern.slf4j.Slf4j;
import org.apache.gobblin.configuration.ConfigurationKeys;
import org.apache.gobblin.instrumented.Instrumented;
@@ -39,9 +40,9 @@ import org.apache.gobblin.metrics.MetricContext;
import org.apache.gobblin.metrics.ServiceMetricNames;
import org.apache.gobblin.runtime.api.DagActionStore;
import org.apache.gobblin.runtime.api.MultiActiveLeaseArbiter;
-import org.apache.gobblin.runtime.api.MysqlMultiActiveLeaseArbiter;
import org.apache.gobblin.scheduler.JobScheduler;
import org.apache.gobblin.scheduler.SchedulerService;
+import org.apache.gobblin.service.modules.core.GobblinServiceGuiceModule;
import org.apache.gobblin.service.modules.scheduler.GobblinServiceJobScheduler;
import org.apache.gobblin.util.ConfigUtils;
import org.quartz.JobDataMap;
@@ -53,122 +54,92 @@ import org.quartz.impl.JobDetailImpl;
/**
- * Handler used to coordinate multiple hosts with enabled schedulers to respond to flow action events. It uses the
- * {@link MysqlMultiActiveLeaseArbiter} to determine a single lease owner at a given time
- * for a flow action event. After acquiring the lease, it persists the flow action event to the {@link DagActionStore}
- * to be eventually acted upon by the host with the active DagManager. Once it has completed this action, it will mark
- * the lease as completed by calling the
- * {@link MysqlMultiActiveLeaseArbiter#recordLeaseSuccess(MultiActiveLeaseArbiter.LeaseObtainedStatus)} method. Hosts
+ * Handler used to coordinate multiple hosts with enabled schedulers to respond to launch dag action events and ensure
+ * no adhoc or scheduled flow launches are missed. It uses the {@link MultiActiveLeaseArbiter} to determine a single
+ * lease owner at a given time for the launch dag action event. After acquiring the lease, it persists the dag action
+ * event to the {@link DagActionStore} to be eventually acted upon by execution module of host(s) to execute the launch.
+ * Once it has completed persisting the action to the store, it will mark the lease as completed by calling the
+ * {@link MultiActiveLeaseArbiter#recordLeaseSuccess(MultiActiveLeaseArbiter.LeaseObtainedStatus)} method. Hosts
* that do not gain the lease for the event, instead schedule a reminder using the {@link SchedulerService} to check
* back in on the previous lease owner's completion status after the lease should expire to ensure the event is handled
* in failure cases.
*/
@Slf4j
-public class FlowTriggerHandler {
+public class FlowLaunchHandler {
+ private final Optional<MultiActiveLeaseArbiter> multiActiveLeaseArbiter;
+ private Optional<DagActionStore> dagActionStore;
+ private final MetricContext metricContext;
private final int schedulerMaxBackoffMillis;
private static Random random = new Random();
- protected Optional<MultiActiveLeaseArbiter> multiActiveLeaseArbiter;
protected SchedulerService schedulerService;
- protected Optional<DagActionStore> dagActionStore;
- private MetricContext metricContext;
private ContextAwareMeter numFlowsSubmitted;
-
- private ContextAwareCounter leaseObtainedCount;
-
- private ContextAwareCounter leasedToAnotherStatusCount;
-
- private ContextAwareCounter noLongerLeasingStatusCount;
private ContextAwareCounter jobDoesNotExistInSchedulerCount;
private ContextAwareCounter failedToSetEventReminderCount;
- private ContextAwareMeter leasesObtainedDueToReminderCount;
- private ContextAwareMeter failedToRecordLeaseSuccessCount;
- private ContextAwareMeter recordedLeaseSuccessCount;
@Inject
- public FlowTriggerHandler(Config config, Optional<MultiActiveLeaseArbiter> leaseDeterminationStore,
+ public FlowLaunchHandler(Config config, @Named(GobblinServiceGuiceModule.SCHEDULER_LEASE_ARBITER_NAME) Optional<MultiActiveLeaseArbiter> leaseArbiter,
SchedulerService schedulerService, Optional<DagActionStore> dagActionStore) {
+ this.multiActiveLeaseArbiter = leaseArbiter;
+ this.dagActionStore = dagActionStore;
this.schedulerMaxBackoffMillis = ConfigUtils.getInt(config, ConfigurationKeys.SCHEDULER_MAX_BACKOFF_MILLIS_KEY,
ConfigurationKeys.DEFAULT_SCHEDULER_MAX_BACKOFF_MILLIS);
- this.multiActiveLeaseArbiter = leaseDeterminationStore;
this.schedulerService = schedulerService;
- this.dagActionStore = dagActionStore;
+
+ // Initialize FlowLaunchHandler related metrics
this.metricContext = Instrumented.getMetricContext(new org.apache.gobblin.configuration.State(ConfigUtils.configToProperties(config)),
this.getClass());
this.numFlowsSubmitted = metricContext.contextAwareMeter(ServiceMetricNames.GOBBLIN_FLOW_TRIGGER_HANDLER_NUM_FLOWS_SUBMITTED);
- this.leaseObtainedCount = this.metricContext.contextAwareCounter(ServiceMetricNames.FLOW_TRIGGER_HANDLER_LEASE_OBTAINED_COUNT);
- this.leasedToAnotherStatusCount = this.metricContext.contextAwareCounter(ServiceMetricNames.FLOW_TRIGGER_HANDLER_LEASED_TO_ANOTHER_COUNT);
- this.noLongerLeasingStatusCount = this.metricContext.contextAwareCounter(ServiceMetricNames.FLOW_TRIGGER_HANDLER_NO_LONGER_LEASING_COUNT);
- this.jobDoesNotExistInSchedulerCount = this.metricContext.contextAwareCounter(ServiceMetricNames.FLOW_TRIGGER_HANDLER_JOB_DOES_NOT_EXIST_COUNT);
- this.failedToSetEventReminderCount = this.metricContext.contextAwareCounter(ServiceMetricNames.FLOW_TRIGGER_HANDLER_FAILED_TO_SET_REMINDER_COUNT);
- this.leasesObtainedDueToReminderCount = this.metricContext.contextAwareMeter(ServiceMetricNames.FLOW_TRIGGER_HANDLER_LEASES_OBTAINED_DUE_TO_REMINDER_COUNT);
- this.failedToRecordLeaseSuccessCount = this.metricContext.contextAwareMeter(ServiceMetricNames.FLOW_TRIGGER_HANDLER_FAILED_TO_RECORD_LEASE_SUCCESS_COUNT);
- this.recordedLeaseSuccessCount = this.metricContext.contextAwareMeter(ServiceMetricNames.FLOW_TRIGGER_HANDLER_RECORDED_LEASE_SUCCESS_COUNT);
+ this.jobDoesNotExistInSchedulerCount = metricContext.contextAwareCounter(ServiceMetricNames.FLOW_TRIGGER_HANDLER_JOB_DOES_NOT_EXIST_COUNT);
+ this.failedToSetEventReminderCount = metricContext.contextAwareCounter(ServiceMetricNames.FLOW_TRIGGER_HANDLER_FAILED_TO_SET_REMINDER_COUNT);
}
/**
- * This method is used in the multi-active scheduler case for one or more hosts to respond to a flow action event
- * by attempting a lease for the flow event and processing the result depending on the status of the attempt.
+ * This method is used in the multi-active scheduler case for one or more hosts to respond to a launch dag action
+ * event triggered by the scheduler by attempting a lease for the launch event and processing the result depending on
+ * the status of the attempt.
* @param jobProps
- * @param flowAction
+ * @param dagAction
* @param eventTimeMillis
* @param isReminderEvent
* @param skipFlowExecutionIdReplacement
* @throws IOException
*/
- public void handleTriggerEvent(Properties jobProps, DagActionStore.DagAction flowAction, long eventTimeMillis,
- boolean isReminderEvent, boolean skipFlowExecutionIdReplacement) throws IOException {
- if (multiActiveLeaseArbiter.isPresent()) {
- MultiActiveLeaseArbiter.LeaseAttemptStatus leaseAttemptStatus = multiActiveLeaseArbiter.get().tryAcquireLease(
- flowAction, eventTimeMillis, isReminderEvent, skipFlowExecutionIdReplacement);
- // The flow action contained in the`LeaseAttemptStatus` from the lease arbiter contains an updated flow execution
- // id. From this point onwards, always use the newer version of the flow action to easily track the action through
- // orchestration and execution.
- if (leaseAttemptStatus instanceof MultiActiveLeaseArbiter.LeaseObtainedStatus) {
- if (isReminderEvent) {
- this.leasesObtainedDueToReminderCount.mark();
+ public void handleFlowLaunchTriggerEvent(Properties jobProps, DagActionStore.DagAction dagAction,
+ long eventTimeMillis, boolean isReminderEvent, boolean skipFlowExecutionIdReplacement) throws IOException {
+ if (this.multiActiveLeaseArbiter.isPresent()) {
+ MultiActiveLeaseArbiter.LeaseAttemptStatus
+ leaseAttemptStatus = this.multiActiveLeaseArbiter.get()
+ .tryAcquireLease(dagAction, eventTimeMillis, isReminderEvent, skipFlowExecutionIdReplacement);
+ if (leaseAttemptStatus instanceof MultiActiveLeaseArbiter.LeaseObtainedStatus) {
+ MultiActiveLeaseArbiter.LeaseObtainedStatus leaseObtainedStatus =
+ (MultiActiveLeaseArbiter.LeaseObtainedStatus) leaseAttemptStatus;
+ if (persistDagAction(leaseObtainedStatus)) {
+ log.info("Successfully persisted lease: [{}, eventTimestamp: {}] ", leaseObtainedStatus.getDagAction(),
+ leaseObtainedStatus.getEventTimeMillis());
+ return;
+ }
+ // If persisting the dag action failed, then we set another trigger for this event to occur immediately to
+ // re-attempt handling the event
+ scheduleReminderForEvent(jobProps,
+ new MultiActiveLeaseArbiter.LeasedToAnotherStatus(leaseObtainedStatus.getDagAction(), 0L), eventTimeMillis);
+ } else if (leaseAttemptStatus instanceof MultiActiveLeaseArbiter.LeasedToAnotherStatus) {
+ scheduleReminderForEvent(jobProps, (MultiActiveLeaseArbiter.LeasedToAnotherStatus) leaseAttemptStatus,
+ eventTimeMillis);
}
- MultiActiveLeaseArbiter.LeaseObtainedStatus leaseObtainedStatus = (MultiActiveLeaseArbiter.LeaseObtainedStatus)
- leaseAttemptStatus;
- this.leaseObtainedCount.inc();
- if (persistFlowAction(leaseObtainedStatus)) {
- log.info("Successfully persisted lease: [{}, eventTimestamp: {}] ", leaseObtainedStatus.getFlowAction(),
- leaseObtainedStatus.getEventTimeMillis());
- this.recordedLeaseSuccessCount.mark();
- return;
- }
- this.failedToRecordLeaseSuccessCount.mark();
- // If persisting the flow action failed, then we set another trigger for this event to occur immediately to
- // re-attempt handling the event
- scheduleReminderForEvent(jobProps,
- new MultiActiveLeaseArbiter.LeasedToAnotherStatus(leaseObtainedStatus.getFlowAction(),
- 0L), eventTimeMillis);
- return;
- } else if (leaseAttemptStatus instanceof MultiActiveLeaseArbiter.LeasedToAnotherStatus) {
- this.leasedToAnotherStatusCount.inc();
- scheduleReminderForEvent(jobProps,
- (MultiActiveLeaseArbiter.LeasedToAnotherStatus) leaseAttemptStatus, eventTimeMillis);
- return;
- } else if (leaseAttemptStatus instanceof MultiActiveLeaseArbiter.NoLongerLeasingStatus) {
- this.noLongerLeasingStatusCount.inc();
- log.debug("Received type of leaseAttemptStatus: [{}, eventTimestamp: {}] ", leaseAttemptStatus.getClass().getName(),
- eventTimeMillis);
- return;
+ // Otherwise leaseAttemptStatus instanceof MultiActiveLeaseArbiter.NoLongerLeasingStatus & no need to do anything
+ } else {
+ throw new RuntimeException(String.format("Multi-active scheduler is not enabled so trigger event should not be "
+ + "handled with this method."));
}
- throw new RuntimeException(String.format("Received type of leaseAttemptStatus: %s not handled by this method",
- leaseAttemptStatus.getClass().getName()));
- } else {
- throw new RuntimeException(String.format("Multi-active scheduler is not enabled so trigger event should not be "
- + "handled with this method."));
- }
}
- // Called after obtaining a lease to persist the flow action to {@link DagActionStore} and mark the lease as done
- private boolean persistFlowAction(MultiActiveLeaseArbiter.LeaseObtainedStatus leaseStatus) {
+ // Called after obtaining a lease to persist the dag action to {@link DagActionStore} and mark the lease as done
+ private boolean persistDagAction(MultiActiveLeaseArbiter.LeaseObtainedStatus leaseStatus) {
if (this.dagActionStore.isPresent() && this.multiActiveLeaseArbiter.isPresent()) {
try {
- DagActionStore.DagAction flowAction = leaseStatus.getFlowAction();
- this.dagActionStore.get().addDagAction(flowAction.getFlowGroup(), flowAction.getFlowName(), flowAction.getFlowExecutionId(), flowAction.getFlowActionType());
- // If the flow action has been persisted to the {@link DagActionStore} we can close the lease
+ DagActionStore.DagAction dagAction = leaseStatus.getDagAction();
+ this.dagActionStore.get().addDagAction(dagAction.getFlowGroup(), dagAction.getFlowName(), dagAction.getFlowExecutionId(), dagAction.getDagActionType());
+ // If the dag action has been persisted to the {@link DagActionStore} we can close the lease
this.numFlowsSubmitted.mark();
return this.multiActiveLeaseArbiter.get().recordLeaseSuccess(leaseStatus);
} catch (IOException e) {
@@ -182,15 +153,15 @@ public class FlowTriggerHandler {
}
/**
- * This method is used by {@link FlowTriggerHandler.handleTriggerEvent} to schedule a self-reminder to check on
- * the other participant's progress to finish acting on a flow action after the time the lease should expire.
+ * This method is used by {@link FlowLaunchHandler#handleFlowLaunchTriggerEvent} to schedule a self-reminder to check on
+ * the other participant's progress to finish acting on a dag action after the time the lease should expire.
* @param jobProps
* @param status used to extract event to be reminded for and the minimum time after which reminder should occur
* @param triggerEventTimeMillis the event timestamp we were originally handling
*/
private void scheduleReminderForEvent(Properties jobProps, MultiActiveLeaseArbiter.LeasedToAnotherStatus status,
long triggerEventTimeMillis) {
- DagActionStore.DagAction flowAction = status.getFlowAction();
+ DagActionStore.DagAction dagAction = status.getDagAction();
JobKey origJobKey = new JobKey(jobProps.getProperty(ConfigurationKeys.JOB_NAME_KEY, "<<no job name>>"),
jobProps.getProperty(ConfigurationKeys.JOB_GROUP_KEY, "<<no job group>>"));
try {
@@ -200,8 +171,8 @@ public class FlowTriggerHandler {
return;
}
Trigger reminderTrigger = createAndScheduleReminder(origJobKey, status, triggerEventTimeMillis);
- log.info("Flow Trigger Handler - [{}, eventTimestamp: {}] - SCHEDULED REMINDER for event {} in {} millis",
- flowAction, triggerEventTimeMillis, status.getEventTimeMillis(), reminderTrigger.getNextFireTime());
+ log.info("Flow Launch Handler - [{}, eventTimestamp: {}] - SCHEDULED REMINDER for event {} in {} millis",
+ dagAction, triggerEventTimeMillis, status.getEventTimeMillis(), reminderTrigger.getNextFireTime());
} catch (SchedulerException e) {
log.warn("Failed to add job reminder due to SchedulerException for job {} trigger event {}. Exception: {}",
origJobKey, status.getEventTimeMillis(), e);
@@ -229,8 +200,8 @@ public class FlowTriggerHandler {
JobDetailImpl jobDetail = createJobDetailForReminderEvent(origJobKey, reminderJobKey, status);
Trigger reminderTrigger = JobScheduler.createTriggerForJob(reminderJobKey, getJobPropertiesFromJobDetail(jobDetail),
Optional.of(reminderSuffix));
- log.debug("Flow Trigger Handler - [{}, eventTimestamp: {}] - attempting to schedule reminder for event {} with "
- + "reminderJobKey {} and reminderTriggerKey {}", status.getFlowAction(), triggerEventTimeMillis,
+ log.debug("Flow Launch Handler - [{}, eventTimestamp: {}] - attempting to schedule reminder for event {} with "
+ + "reminderJobKey {} and reminderTriggerKey {}", status.getDagAction(), triggerEventTimeMillis,
status.getEventTimeMillis(), reminderJobKey, reminderTrigger.getKey());
this.schedulerService.getScheduler().scheduleJob(jobDetail, reminderTrigger);
return reminderTrigger;
diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/InstrumentedLeaseArbiter.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/InstrumentedLeaseArbiter.java
new file mode 100644
index 000000000..f06def5b1
--- /dev/null
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/InstrumentedLeaseArbiter.java
@@ -0,0 +1,112 @@
+/*
+ * 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 com.typesafe.config.Config;
+
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.ContextAwareMeter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.runtime.api.DagActionStore;
+import org.apache.gobblin.runtime.api.MultiActiveLeaseArbiter;
+import org.apache.gobblin.util.ConfigUtils;
+
+
+/*
+ A {@link MultiActiveLeaseArbiter} decorator for tracking metrics about lease attempts. The metrics can be used to
+ compare relative performance, such as collisions/contention, among arbitration participants.
+ */
+@Slf4j
+public class InstrumentedLeaseArbiter implements MultiActiveLeaseArbiter {
+ private final MultiActiveLeaseArbiter decoratedMultiActiveLeaseArbiter;
+ @Getter
+ protected MetricContext metricContext;
+ private ContextAwareCounter leaseObtainedCount;
+
+ private ContextAwareCounter leasedToAnotherStatusCount;
+
+ private ContextAwareCounter noLongerLeasingStatusCount;
+ private ContextAwareMeter leasesObtainedDueToReminderCount;
+ private ContextAwareMeter recordedLeaseSuccessCount;
+ private ContextAwareMeter failedToRecordLeaseSuccessCount;
+
+ public InstrumentedLeaseArbiter(Config config, MultiActiveLeaseArbiter leaseDeterminationStore,
+ String metricsPrefix) {
+ this.decoratedMultiActiveLeaseArbiter = leaseDeterminationStore;
+ this.metricContext = Instrumented.getMetricContext(new org.apache.gobblin.configuration.State(ConfigUtils.configToProperties(config)),
+ this.getClass());
+ initializeMetrics(metricsPrefix);
+ }
+
+ private void initializeMetrics(String metricsPrefix) {
+ // If a valid metrics prefix is provided then add a delimiter after it
+ if (!metricsPrefix.isEmpty()) {
+ metricsPrefix = metricsPrefix.concat(".");
+ }
+ this.leaseObtainedCount = this.metricContext.contextAwareCounter(metricsPrefix + ServiceMetricNames.FLOW_TRIGGER_HANDLER_LEASE_OBTAINED_COUNT);
+ this.leasedToAnotherStatusCount = this.metricContext.contextAwareCounter(metricsPrefix + ServiceMetricNames.FLOW_TRIGGER_HANDLER_LEASED_TO_ANOTHER_COUNT);
+ this.noLongerLeasingStatusCount = this.metricContext.contextAwareCounter(metricsPrefix + ServiceMetricNames.FLOW_TRIGGER_HANDLER_NO_LONGER_LEASING_COUNT);
+ this.leasesObtainedDueToReminderCount = this.metricContext.contextAwareMeter(metricsPrefix + ServiceMetricNames.FLOW_TRIGGER_HANDLER_LEASES_OBTAINED_DUE_TO_REMINDER_COUNT);
+ this.recordedLeaseSuccessCount = metricContext.contextAwareMeter(ServiceMetricNames.FLOW_TRIGGER_HANDLER_RECORDED_LEASE_SUCCESS_COUNT);
+ this.failedToRecordLeaseSuccessCount = metricContext.contextAwareMeter(ServiceMetricNames.FLOW_TRIGGER_HANDLER_FAILED_TO_RECORD_LEASE_SUCCESS_COUNT);
+ }
+
+ @Override
+ public MultiActiveLeaseArbiter.LeaseAttemptStatus tryAcquireLease(DagActionStore.DagAction dagAction, long eventTimeMillis,
+ boolean isReminderEvent, boolean skipFlowExecutionIdReplacement) throws IOException {
+
+ MultiActiveLeaseArbiter.LeaseAttemptStatus leaseAttemptStatus =
+ decoratedMultiActiveLeaseArbiter.tryAcquireLease(dagAction, eventTimeMillis, isReminderEvent,
+ skipFlowExecutionIdReplacement);
+ log.info("Multi-active scheduler lease attempt for dagAction: {} received type of leaseAttemptStatus: [{}, "
+ + "eventTimestamp: {}] ", dagAction, leaseAttemptStatus.getClass().getName(), eventTimeMillis);
+ if (leaseAttemptStatus instanceof MultiActiveLeaseArbiter.LeaseObtainedStatus) {
+ if (isReminderEvent) {
+ this.leasesObtainedDueToReminderCount.mark();
+ }
+ this.leaseObtainedCount.inc();
+ return leaseAttemptStatus;
+ } else if (leaseAttemptStatus instanceof MultiActiveLeaseArbiter.LeasedToAnotherStatus) {
+ this.leasedToAnotherStatusCount.inc();
+ return leaseAttemptStatus;
+ } else if (leaseAttemptStatus instanceof MultiActiveLeaseArbiter.NoLongerLeasingStatus) {
+ this.noLongerLeasingStatusCount.inc();
+ return leaseAttemptStatus;
+ }
+ throw new RuntimeException(String.format("Received type of leaseAttemptStatus: %s not handled by this method",
+ leaseAttemptStatus.getClass().getName()));
+ }
+
+ @Override
+ public boolean recordLeaseSuccess(LeaseObtainedStatus status)
+ throws IOException {
+ if (this.decoratedMultiActiveLeaseArbiter.recordLeaseSuccess(status)) {
+ this.recordedLeaseSuccessCount.mark();
+ return true;
+ }
+ this.failedToRecordLeaseSuccessCount.mark();
+ return false;
+ }
+}
diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java
index daf6aff3f..14805ee0d 100644
--- a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java
@@ -106,22 +106,23 @@ public class Orchestrator implements SpecCatalogListener, Instrumentable {
private UserQuotaManager quotaManager;
private final FlowCompilationValidationHelper flowCompilationValidationHelper;
- private Optional<FlowTriggerHandler> flowTriggerHandler;
+ private Optional<FlowLaunchHandler> flowTriggerDecorator;
private Optional<FlowCatalog> flowCatalog;
@Getter
private final SharedFlowMetricsSingleton sharedFlowMetricsSingleton;
@Inject
public Orchestrator(Config config, TopologyCatalog topologyCatalog, DagManager dagManager,
- Optional<Logger> log, FlowStatusGenerator flowStatusGenerator, Optional<FlowTriggerHandler> flowTriggerHandler,
+ Optional<Logger> log, FlowStatusGenerator flowStatusGenerator, Optional<FlowLaunchHandler> flowTriggerDecorator,
SharedFlowMetricsSingleton sharedFlowMetricsSingleton, Optional<FlowCatalog> flowCatalog,
- DagManagementStateStore dagManagementStateStore, FlowCompilationValidationHelper flowCompilationValidationHelper) throws IOException {
+ Optional<DagManagementStateStore> dagManagementStateStore,
+ FlowCompilationValidationHelper flowCompilationValidationHelper) throws IOException {
_log = log.isPresent() ? log.get() : LoggerFactory.getLogger(getClass());
ClassAliasResolver<SpecCompiler> aliasResolver = new ClassAliasResolver<>(SpecCompiler.class);
this.topologyCatalog = topologyCatalog;
this.dagManager = dagManager;
this.flowStatusGenerator = flowStatusGenerator;
- this.flowTriggerHandler = flowTriggerHandler;
+ this.flowTriggerDecorator = flowTriggerDecorator;
this.sharedFlowMetricsSingleton = sharedFlowMetricsSingleton;
this.flowCatalog = flowCatalog;
try {
@@ -139,7 +140,9 @@ public class Orchestrator implements SpecCatalogListener, Instrumentable {
//At this point, the TopologySpecMap is initialized by the SpecCompiler. Pass the TopologySpecMap to the DagManager.
this.dagManager.setTopologySpecMap(getSpecCompiler().getTopologySpecMap());
- ((MostlyMySqlDagManagementStateStore) dagManagementStateStore).setTopologySpecMap(getSpecCompiler().getTopologySpecMap());
+ if (dagManagementStateStore.isPresent()) {
+ ((MostlyMySqlDagManagementStateStore) dagManagementStateStore.get()).setTopologySpecMap(getSpecCompiler().getTopologySpecMap());
+ }
this.metricContext = Instrumented.getMetricContext(ConfigUtils.configToState(config), this.specCompiler.getClass());
this.flowOrchestrationSuccessFulMeter = this.metricContext.meter(ServiceMetricNames.FLOW_ORCHESTRATION_SUCCESSFUL_METER);
@@ -230,18 +233,19 @@ public class Orchestrator implements SpecCatalogListener, Instrumentable {
Map<String, String> flowMetadata = TimingEventUtils.getFlowMetadata(flowSpec);
String flowExecutionId = String.valueOf(FlowUtils.getOrCreateFlowExecutionId(flowSpec));
- DagActionStore.DagAction flowAction =
- new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId, DagActionStore.FlowActionType.LAUNCH);
+
+ DagActionStore.DagAction launchDagAction =
+ new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId, "", DagActionStore.DagActionType.LAUNCH);
// If multi-active scheduler is enabled do not pass onto DagManager, otherwise scheduler forwards it directly
// Skip flow compilation as well, since we recompile after receiving event from DagActionStoreChangeMonitor later
- if (flowTriggerHandler.isPresent()) {
+ if (flowTriggerDecorator.isPresent()) {
// Adopt consensus flowExecutionId for scheduled flows
- flowTriggerHandler.get().handleTriggerEvent(jobProps, flowAction, triggerTimestampMillis, isReminderEvent,
+ flowTriggerDecorator.get().handleFlowLaunchTriggerEvent(jobProps, launchDagAction, triggerTimestampMillis, isReminderEvent,
flowSpec.isScheduled());
_log.info("Multi-active scheduler finished handling trigger event: [{}, is: {}, triggerEventTimestamp: {}]",
- flowAction, isReminderEvent ? "reminder" : "original", triggerTimestampMillis);
+ launchDagAction, isReminderEvent ? "reminder" : "original", triggerTimestampMillis);
} else {
TimingEvent flowCompilationTimer = new TimingEvent(this.eventSubmitter, TimingEvent.FlowTimings.FLOW_COMPILED);
Optional<Dag<JobExecutionPlan>> compiledDagOptional =
diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/ReminderSettingDagProcLeaseArbiter.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/ReminderSettingDagProcLeaseArbiter.java
new file mode 100644
index 000000000..bc24cd0e7
--- /dev/null
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/ReminderSettingDagProcLeaseArbiter.java
@@ -0,0 +1,199 @@
+/*
+ * 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.Date;
+import java.util.Optional;
+
+import org.quartz.Job;
+import org.quartz.JobBuilder;
+import org.quartz.JobDataMap;
+import org.quartz.JobDetail;
+import org.quartz.JobExecutionContext;
+import org.quartz.SchedulerException;
+import org.quartz.Trigger;
+import org.quartz.TriggerBuilder;
+
+import com.typesafe.config.Config;
+
+import javax.inject.Inject;
+import javax.inject.Named;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.runtime.api.DagActionStore;
+import org.apache.gobblin.runtime.api.MultiActiveLeaseArbiter;
+import org.apache.gobblin.service.modules.core.GobblinServiceGuiceModule;
+
+
+/**
+ * Decorator used to coordinate multiple hosts with execution components enabled to respond to flow action events with
+ * added capabilities to properly handle the result of attempted ownership over these flow action events. It uses the
+ * {@link MultiActiveLeaseArbiter} to determine a single lease owner at a given event time for a flow action event.
+ * If the status of the lease ownership attempt is anything other than an indication the lease has been completed
+ * ({@link org.apache.gobblin.runtime.api.MultiActiveLeaseArbiter.NoLongerLeasingStatus}) then the
+ * {@link MultiActiveLeaseArbiter#tryAcquireLease} method will set a reminder for the flow action using
+ * {@link DagActionReminderScheduler} to reattempt the lease after the current lease holder's grant would have expired.
+ */
+@Slf4j
+public class ReminderSettingDagProcLeaseArbiter implements MultiActiveLeaseArbiter {
+ private final Optional<MultiActiveLeaseArbiter> decoratedLeaseArbiter;
+ private final Optional<DagActionReminderScheduler> dagActionReminderScheduler;
+ private final Config config;
+ private final String MISSING_OPTIONAL_ERROR_MESSAGE = String.format("Multi-active execution is not enabled so dag "
+ + "action should not passed to %s", ReminderSettingDagProcLeaseArbiter.class.getSimpleName());
+
+ @Inject
+ public ReminderSettingDagProcLeaseArbiter(Config config,
+ @Named(GobblinServiceGuiceModule.EXECUTOR_LEASE_ARBITER_NAME) Optional<MultiActiveLeaseArbiter> leaseArbiter,
+ Optional<DagActionReminderScheduler> dagActionReminderScheduler) {
+ this.decoratedLeaseArbiter = leaseArbiter;
+ this.dagActionReminderScheduler = dagActionReminderScheduler;
+ this.config = config;
+ }
+
+ /**
+ * Attempts a lease for a particular job event and sets a reminder to revisit if the lease has not been completed.
+ */
+ @Override
+ public MultiActiveLeaseArbiter.LeaseAttemptStatus tryAcquireLease(DagActionStore.DagAction dagAction, long eventTimeMillis,
+ boolean isReminderEvent, boolean skipFlowExecutionIdReplacement) {
+ if (this.decoratedLeaseArbiter.isPresent()) {
+ try {
+ MultiActiveLeaseArbiter.LeaseAttemptStatus leaseAttemptStatus =
+ this.decoratedLeaseArbiter.get().tryAcquireLease(dagAction, eventTimeMillis, isReminderEvent,
+ skipFlowExecutionIdReplacement);
+ /* Schedule a reminder for the event unless the lease has been completed to safeguard against the case where even
+ we, when we might become the lease owner still fail to complete processing
+ */
+ if (!(leaseAttemptStatus instanceof NoLongerLeasingStatus)) {
+ scheduleReminderForEvent(leaseAttemptStatus);
+ }
+ return leaseAttemptStatus;
+ } catch (SchedulerException | IOException e) {
+ throw new RuntimeException(e);
+ }
+ } else {
+ throw new RuntimeException(MISSING_OPTIONAL_ERROR_MESSAGE);
+ }
+ }
+
+ @Override
+ public boolean recordLeaseSuccess(LeaseObtainedStatus status)
+ throws IOException {
+ if (!this.decoratedLeaseArbiter.isPresent()) {
+ throw new RuntimeException(MISSING_OPTIONAL_ERROR_MESSAGE);
+ }
+ return this.decoratedLeaseArbiter.get().recordLeaseSuccess(status);
+ }
+
+ protected void scheduleReminderForEvent(MultiActiveLeaseArbiter.LeaseAttemptStatus leaseStatus)
+ throws SchedulerException {
+ if (!this.dagActionReminderScheduler.isPresent()) {
+ throw new RuntimeException(MISSING_OPTIONAL_ERROR_MESSAGE);
+ }
+ dagActionReminderScheduler.get().scheduleReminder(leaseStatus.getDagAction(),
+ leaseStatus.getMinimumLingerDurationMillis());
+ }
+
+ /**
+ * Static class used to store information regarding a pending dagAction that needs to be revisited at a later time
+ * by {@link DagManagement} interface to re-attempt a lease on if it has not been completed by the previous owner.
+ * These jobs are scheduled and used by the {@link DagActionReminderScheduler}.
+ */
+ @Slf4j
+ public static class ReminderJob implements Job {
+ public static final String FLOW_ACTION_TYPE_KEY = "flow.actionType";
+ public static final String DAG_MANAGEMENT_KEY = "dag.management";
+
+ @Override
+ public void execute(JobExecutionContext context) {
+ // Get properties from the trigger to create a dagAction
+ JobDataMap jobDataMap = context.getTrigger().getJobDataMap();
+ String flowName = jobDataMap.getString(ConfigurationKeys.FLOW_NAME_KEY);
+ String flowGroup = jobDataMap.getString(ConfigurationKeys.FLOW_GROUP_KEY);
+ String jobName = jobDataMap.getString(ConfigurationKeys.JOB_NAME_KEY);
+ String flowId = jobDataMap.getString(ConfigurationKeys.FLOW_EXECUTION_ID_KEY);
+ DagActionStore.DagActionType dagActionType = DagActionStore.DagActionType.valueOf(
+ jobDataMap.getString(FLOW_ACTION_TYPE_KEY));
+ DagManagement dagManagement = (DagManagement) jobDataMap.get(DAG_MANAGEMENT_KEY);
+
+ log.info("DagProc reminder triggered for (flowGroup: " + flowGroup + ", flowName: " + flowName
+ + ", flowExecutionId: " + flowId + ", jobName: " + jobName +")");
+
+ DagActionStore.DagAction dagAction = new DagActionStore.DagAction(flowGroup, flowName, flowId, jobName,
+ dagActionType);
+
+ try {
+ dagManagement.addDagAction(dagAction);
+ } catch (IOException e) {
+ log.error("Failed to add DagAction to DagManagement. Action: {}", dagAction);
+ }
+ }
+ }
+
+ /**
+ * Creates a key for the reminder job by concatenating all dagAction fields
+ */
+ public static String createDagActionReminderKey(DagActionStore.DagAction dagAction) {
+ return createDagActionReminderKey(dagAction.getFlowName(), dagAction.getFlowGroup(), dagAction.getFlowExecutionId(),
+ dagAction.getJobName(), dagAction.getDagActionType());
+ }
+
+ /**
+ * Creates a key for the reminder job by concatenating flowName, flowGroup, flowExecutionId, jobName, dagActionType
+ * in that order
+ */
+ public static String createDagActionReminderKey(String flowName, String flowGroup, String flowId, String jobName,
+ DagActionStore.DagActionType dagActionType) {
+ return String.format("%s.%s.%s.%s.%s", flowGroup, flowName, flowId, jobName, dagActionType);
+ }
+
+ /**
+ * Creates a jobDetail containing flow and job identifying information in the jobDataMap, uniquely identified
+ * by a key comprised of the dagAction's fields. It also serializes a reference to the {@link DagManagement} object
+ * to be referenced when the trigger fires.
+ */
+ public static JobDetail createReminderJobDetail(DagManagement dagManagement, DagActionStore.DagAction dagAction) {
+ JobDataMap dataMap = new JobDataMap();
+ dataMap.put(ReminderJob.DAG_MANAGEMENT_KEY, dagManagement);
+ dataMap.put(ConfigurationKeys.FLOW_NAME_KEY, dagAction.getFlowName());
+ dataMap.put(ConfigurationKeys.FLOW_GROUP_KEY, dagAction.getFlowGroup());
+ dataMap.put(ConfigurationKeys.JOB_NAME_KEY, dagAction.getJobName());
+ dataMap.put(ConfigurationKeys.FLOW_EXECUTION_ID_KEY, dagAction.getFlowExecutionId());
+ dataMap.put(ReminderJob.FLOW_ACTION_TYPE_KEY, dagAction.getDagActionType());
+
+ return JobBuilder.newJob(ReminderJob.class)
+ .withIdentity(createDagActionReminderKey(dagAction), dagAction.getFlowName())
+ .usingJobData(dataMap)
+ .build();
+ }
+
+ /**
+ * Creates a Trigger object with the same key as the ReminderJob (since only one trigger is expected to be associated
+ * with a job at any given time) that should fire after `reminderDurationMillis` millis.
+ */
+ public static Trigger createReminderJobTrigger(DagActionStore.DagAction dagAction, long reminderDurationMillis) {
+ Trigger trigger = TriggerBuilder.newTrigger()
+ .withIdentity(createDagActionReminderKey(dagAction), dagAction.getFlowName())
+ .startAt(new Date(System.currentTimeMillis() + reminderDurationMillis))
+ .build();
+ return trigger;
+ }
+}
diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/LaunchDagProc.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/LaunchDagProc.java
index fec15851c..3e10b8319 100644
--- a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/LaunchDagProc.java
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/LaunchDagProc.java
@@ -192,7 +192,6 @@ public class LaunchDagProc extends DagProc<Optional<Dag<JobExecutionPlan>>, Opti
}
}
- @Override
protected void sendNotification(Optional<Dag<JobExecutionPlan>> result, EventSubmitter eventSubmitter) {
}
}
diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/DagTask.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/DagTask.java
index 114864e64..a77f3cfcf 100644
--- a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/DagTask.java
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/DagTask.java
@@ -56,5 +56,12 @@ public abstract class DagTask {
* Returns true if concluding dag task finished successfully otherwise false.
*/
// todo call it from the right place
- public abstract boolean conclude() throws IOException;
+ public boolean conclude() {
+ try {
+ return this.leaseObtainedStatus.completeLease();
+ } catch (IOException e) {
+ // TODO: Decide appropriate exception to throw and add to the commit method's signature
+ throw new RuntimeException(e);
+ }
+ }
}
diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/LaunchDagTask.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/LaunchDagTask.java
index 5093eaf3e..c83edacea 100644
--- a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/LaunchDagTask.java
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/LaunchDagTask.java
@@ -34,10 +34,4 @@ public class LaunchDagTask extends DagTask {
public <T> T host(DagTaskVisitor<T> visitor) {
return visitor.meet(this);
}
-
- @Override
- public boolean conclude() {
- // todo - release lease
- return true;
- }
}
diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/restli/GobblinServiceFlowExecutionResourceHandlerWithWarmStandby.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/restli/GobblinServiceFlowExecutionResourceHandlerWithWarmStandby.java
index f391e8fe7..0b989c3b1 100644
--- a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/restli/GobblinServiceFlowExecutionResourceHandlerWithWarmStandby.java
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/restli/GobblinServiceFlowExecutionResourceHandlerWithWarmStandby.java
@@ -51,18 +51,18 @@ public class GobblinServiceFlowExecutionResourceHandlerWithWarmStandby extends G
@Override
public void resume(ComplexResourceKey<FlowStatusId, EmptyRecord> key) {
FlowStatusId id = key.getKey();
- addDagAction(id.getFlowGroup(), id.getFlowName(), id.getFlowExecutionId(), DagActionStore.FlowActionType.RESUME);
+ addDagAction(id.getFlowGroup(), id.getFlowName(), id.getFlowExecutionId(), DagActionStore.DagActionType.RESUME);
}
@Override
public UpdateResponse delete(ComplexResourceKey<org.apache.gobblin.service.FlowStatusId, EmptyRecord> key) {
FlowStatusId id = key.getKey();
- addDagAction(id.getFlowGroup(), id.getFlowName(), id.getFlowExecutionId(), DagActionStore.FlowActionType.KILL);
+ addDagAction(id.getFlowGroup(), id.getFlowName(), id.getFlowExecutionId(), DagActionStore.DagActionType.KILL);
return new UpdateResponse(HttpStatus.S_200_OK);
}
/** NOTE: may throw {@link RestLiServiceException}, see: https://linkedin.github.io/rest.li/user_guide/restli_server#returning-errors */
- protected void addDagAction(String flowGroup, String flowName, Long flowExecutionId, DagActionStore.FlowActionType actionType) {
+ protected void addDagAction(String flowGroup, String flowName, Long flowExecutionId, DagActionStore.DagActionType actionType) {
try {
// If an existing resume request is still pending then do not accept this request
if (this.dagActionStore.exists(flowGroup, flowName, flowExecutionId.toString(), actionType)) {
diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobScheduler.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobScheduler.java
index 59d7b0c9b..4b47264f4 100644
--- a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobScheduler.java
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobScheduler.java
@@ -86,7 +86,7 @@ import org.apache.gobblin.service.ServiceConfigKeys;
import org.apache.gobblin.service.modules.flowgraph.Dag;
import org.apache.gobblin.service.modules.orchestration.DagManagementStateStore;
import org.apache.gobblin.service.modules.orchestration.DagManager;
-import org.apache.gobblin.service.modules.orchestration.FlowTriggerHandler;
+import org.apache.gobblin.service.modules.orchestration.FlowLaunchHandler;
import org.apache.gobblin.service.modules.orchestration.Orchestrator;
import org.apache.gobblin.service.modules.orchestration.UserQuotaManager;
import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
@@ -120,7 +120,7 @@ public class GobblinServiceJobScheduler extends JobScheduler implements SpecCata
protected final Orchestrator orchestrator;
protected final Boolean isWarmStandbyEnabled;
protected final Optional<UserQuotaManager> quotaManager;
- protected final Optional<FlowTriggerHandler> flowTriggerHandler;
+ protected final Optional<FlowLaunchHandler> flowTriggerHandler;
@Getter
protected final Map<String, FlowSpec> scheduledFlowSpecs;
@Getter
@@ -177,7 +177,7 @@ public class GobblinServiceJobScheduler extends JobScheduler implements SpecCata
Optional<HelixManager> helixManager, Optional<FlowCatalog> flowCatalog,
Orchestrator orchestrator, SchedulerService schedulerService, Optional<UserQuotaManager> quotaManager, Optional<Logger> log,
@Named(InjectionNames.WARM_STANDBY_ENABLED) boolean isWarmStandbyEnabled,
- Optional<FlowTriggerHandler> flowTriggerHandler) throws Exception {
+ Optional<FlowLaunchHandler> flowTriggerHandler) throws Exception {
super(ConfigUtils.configToProperties(config), schedulerService);
_log = log.isPresent() ? log.get() : LoggerFactory.getLogger(getClass());
@@ -215,14 +215,14 @@ public class GobblinServiceJobScheduler extends JobScheduler implements SpecCata
public GobblinServiceJobScheduler(String serviceName, Config config, FlowStatusGenerator flowStatusGenerator,
Optional<HelixManager> helixManager, Optional<FlowCatalog> flowCatalog, TopologyCatalog topologyCatalog,
DagManager dagManager, Optional<UserQuotaManager> quotaManager, SchedulerService schedulerService,
- Optional<Logger> log, boolean isWarmStandbyEnabled, Optional <FlowTriggerHandler> flowTriggerHandler,
- SharedFlowMetricsSingleton sharedFlowMetricsSingleton, DagManagementStateStore dagManagementStateStore,
+ Optional<Logger> log, boolean isWarmStandbyEnabled, Optional <FlowLaunchHandler> flowTriggerDecorator,
+ SharedFlowMetricsSingleton sharedFlowMetricsSingleton, Optional<DagManagementStateStore> dagManagementStateStore,
FlowCompilationValidationHelper flowCompilationValidationHelper)
throws Exception {
this(serviceName, config, helixManager, flowCatalog,
- new Orchestrator(config, topologyCatalog, dagManager, log, flowStatusGenerator, flowTriggerHandler,
+ new Orchestrator(config, topologyCatalog, dagManager, log, flowStatusGenerator, flowTriggerDecorator,
sharedFlowMetricsSingleton, flowCatalog, dagManagementStateStore, flowCompilationValidationHelper),
- schedulerService, quotaManager, log, isWarmStandbyEnabled, flowTriggerHandler);
+ schedulerService, quotaManager, log, isWarmStandbyEnabled, flowTriggerDecorator);
}
public synchronized void setActive(boolean isActive) {
diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java b/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java
index 7260e45d2..30c55f4e0 100644
--- a/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java
@@ -197,10 +197,11 @@ public class DagActionStoreChangeMonitor extends HighLevelConsumer {
return;
}
- DagActionStore.FlowActionType dagActionType = DagActionStore.FlowActionType.valueOf(value.getDagAction().toString());
+ DagActionStore.DagActionType dagActionType = DagActionStore.DagActionType.valueOf(value.getDagAction().toString());
// Used to easily log information to identify the dag action
- DagActionStore.DagAction dagAction = new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId,
+ // TODO: add jobName to the dagAction change event
+ DagActionStore.DagAction dagAction = new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId, "",
dagActionType);
// We only expect INSERT and DELETE operations done to this table. INSERTs correspond to any type of
@@ -235,16 +236,16 @@ public class DagActionStoreChangeMonitor extends HighLevelConsumer {
*/
protected void handleDagAction(DagActionStore.DagAction dagAction, boolean isStartup) {
log.info("(" + (isStartup ? "on-startup" : "post-startup") + ") DagAction change ({}) received for flow: {}",
- dagAction.getFlowActionType(), dagAction);
- if (dagAction.getFlowActionType().equals(DagActionStore.FlowActionType.RESUME)) {
+ dagAction.getDagActionType(), dagAction);
+ if (dagAction.getDagActionType().equals(DagActionStore.DagActionType.RESUME)) {
dagManager.handleResumeFlowRequest(dagAction.getFlowGroup(), dagAction.getFlowName(),
Long.parseLong(dagAction.getFlowExecutionId()));
this.resumesInvoked.mark();
- } else if (dagAction.getFlowActionType().equals(DagActionStore.FlowActionType.KILL)) {
+ } else if (dagAction.getDagActionType().equals(DagActionStore.DagActionType.KILL)) {
dagManager.handleKillFlowRequest(dagAction.getFlowGroup(), dagAction.getFlowName(),
Long.parseLong(dagAction.getFlowExecutionId()));
this.killsInvoked.mark();
- } else if (dagAction.getFlowActionType().equals(DagActionStore.FlowActionType.LAUNCH)) {
+ } else if (dagAction.getDagActionType().equals(DagActionStore.DagActionType.LAUNCH)) {
// If multi-active scheduler is NOT turned on we should not receive these type of events
if (!this.isMultiActiveSchedulerEnabled) {
this.unexpectedErrors.mark();
@@ -253,7 +254,7 @@ public class DagActionStoreChangeMonitor extends HighLevelConsumer {
}
submitFlowToDagManagerHelper(dagAction, isStartup);
} else {
- log.warn("Received unsupported dagAction {}. Expected to be a KILL, RESUME, or LAUNCH", dagAction.getFlowActionType());
+ log.warn("Received unsupported dagAction {}. Expected to be a KILL, RESUME, or LAUNCH", dagAction.getDagActionType());
this.unexpectedErrors.mark();
}
}
@@ -272,7 +273,7 @@ public class DagActionStoreChangeMonitor extends HighLevelConsumer {
*/
protected void submitFlowToDagManagerHelper(DagActionStore.DagAction dagAction,
LaunchSubmissionMetricProxy launchSubmissionMetricProxy) {
- Preconditions.checkArgument(dagAction.getFlowActionType() == DagActionStore.FlowActionType.LAUNCH);
+ Preconditions.checkArgument(dagAction.getDagActionType() == DagActionStore.DagActionType.LAUNCH);
log.info("Forward launch flow event to DagManager for action {}", dagAction);
// Retrieve job execution plan by recompiling the flow spec to send to the DagManager
FlowId flowId = dagAction.getFlowId();
diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagManagementDagActionStoreChangeMonitor.java b/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagManagementDagActionStoreChangeMonitor.java
index 6bfc92045..ab3ea9c37 100644
--- a/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagManagementDagActionStoreChangeMonitor.java
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagManagementDagActionStoreChangeMonitor.java
@@ -58,11 +58,11 @@ public class DagManagementDagActionStoreChangeMonitor extends DagActionStoreChan
@Override
protected void handleDagAction(DagActionStore.DagAction dagAction, boolean isStartup) {
log.info("(" + (isStartup ? "on-startup" : "post-startup") + ") DagAction change ({}) received for flow: {}",
- dagAction.getFlowActionType(), dagAction);
+ dagAction.getDagActionType(), dagAction);
LaunchSubmissionMetricProxy launchSubmissionMetricProxy = isStartup ? ON_STARTUP : POST_STARTUP;
try {
// todo - add actions for other other type of dag actions
- if (dagAction.getFlowActionType().equals(DagActionStore.FlowActionType.LAUNCH)) {
+ if (dagAction.getDagActionType().equals(DagActionStore.DagActionType.LAUNCH)) {
// If multi-active scheduler is NOT turned on we should not receive these type of events
if (!this.isMultiActiveSchedulerEnabled) {
this.unexpectedLaunchEventErrors.mark();
@@ -71,7 +71,7 @@ public class DagManagementDagActionStoreChangeMonitor extends DagActionStoreChan
}
dagManagement.addDagAction(dagAction);
} else {
- log.warn("Received unsupported dagAction {}. Expected to be a KILL, RESUME, or LAUNCH", dagAction.getFlowActionType());
+ log.warn("Received unsupported dagAction {}. Expected to be a KILL, RESUME, or LAUNCH", dagAction.getDagActionType());
this.unexpectedErrors.mark();
}
} catch (IOException e) {
diff --git a/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagManagementTaskStreamImplTest.java b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagManagementTaskStreamImplTest.java
index f73055df6..68034658c 100644
--- a/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagManagementTaskStreamImplTest.java
+++ b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagManagementTaskStreamImplTest.java
@@ -17,7 +17,6 @@
package org.apache.gobblin.service.modules.orchestration;
-import java.io.IOException;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;
@@ -34,11 +33,19 @@ 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.DagActionStore;
+import org.apache.gobblin.runtime.api.MultiActiveLeaseArbiter;
import org.apache.gobblin.runtime.api.TopologySpec;
import org.apache.gobblin.service.modules.orchestration.proc.DagProc;
import org.apache.gobblin.service.modules.orchestration.task.DagTask;
import org.apache.gobblin.service.modules.orchestration.task.LaunchDagTask;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+
public class DagManagementTaskStreamImplTest {
private static final String TEST_USER = "testUser";
private static final String TEST_PASSWORD = "testPassword";
@@ -70,18 +77,32 @@ public class DagManagementTaskStreamImplTest {
MostlyMySqlDagManagementStateStore dagManagementStateStore = new MostlyMySqlDagManagementStateStore(config, null, null);
dagManagementStateStore.setTopologySpecMap(topologySpecMap);
this.dagManagementTaskStream =
- new DagManagementTaskStreamImpl(config, Optional.empty());
+ new DagManagementTaskStreamImpl(config, Optional.empty(), Optional.of(mock(ReminderSettingDagProcLeaseArbiter.class)));
this.dagProcFactory = new DagProcFactory(null);
this.dagProcEngineThread = new DagProcessingEngine.DagProcEngineThread(
this.dagManagementTaskStream, this.dagProcFactory, dagManagementStateStore);
}
- // This tests adding and removal of dag actions from dag task stream
- // when we have different dag procs in future, we can test dag processing and exception handling
+ /* This tests adding and removal of dag actions from dag task stream with a launch task. It verifies that the
+ {@link DagManagementTaskStreamImpl#next()} call blocks until a {@link LeaseAttemptStatus.LeaseObtainedStatus} is
+ returned for a particular action.
+ TODO: when we have different dag procs in future, update this test to add other types of actions (and tasks)
+ */
@Test
- public void addRemoveDagActions() throws IOException {
- dagManagementTaskStream.addDagAction(
- new DagActionStore.DagAction("fg", "fn", "12345", DagActionStore.FlowActionType.LAUNCH));
+ public void addRemoveDagActions() {
+ /* Three duplicate actions are added to the task stream, since the first two calls to lease arbitration will return
+ statuses that should cause the next() method to continue polling for tasks before finally providing the
+ LeaseObtainedStatus to the taskStream to break its loop and return a newly created dagTask
+ */
+ DagActionStore.DagAction launchAction = new DagActionStore.DagAction("fg", "fn", "12345", "jn", DagActionStore.DagActionType.LAUNCH);
+ dagManagementTaskStream.addDagAction(launchAction);
+ dagManagementTaskStream.addDagAction(launchAction);
+ dagManagementTaskStream.addDagAction(launchAction);
+ when(dagManagementTaskStream.getReminderSettingDagProcLeaseArbiter().get()
+ .tryAcquireLease(any(DagActionStore.DagAction.class), anyLong(), anyBoolean(), anyBoolean()))
+ .thenReturn(new MultiActiveLeaseArbiter.NoLongerLeasingStatus(),
+ new MultiActiveLeaseArbiter.LeasedToAnotherStatus(launchAction, 15),
+ new MultiActiveLeaseArbiter.LeaseObtainedStatus(launchAction, 0, 5, null));
DagTask dagTask = dagManagementTaskStream.next();
Assert.assertTrue(dagTask instanceof LaunchDagTask);
DagProc dagProc = dagTask.host(this.dagProcFactory);
diff --git a/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagManagerFlowTest.java b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagManagerFlowTest.java
index 322cf804e..ef7296393 100644
--- a/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagManagerFlowTest.java
+++ b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagManagerFlowTest.java
@@ -87,8 +87,8 @@ public class DagManagerFlowTest {
.build();
dagActionStore = new MysqlDagActionStore(config);
- dagActionStore.addDagAction(flowGroup, flowName, flowExecutionId, DagActionStore.FlowActionType.KILL);
- dagActionStore.addDagAction(flowGroup, flowName, flowExecutionId_2, DagActionStore.FlowActionType.RESUME);
+ dagActionStore.addDagAction(flowGroup, flowName, flowExecutionId, DagActionStore.DagActionType.KILL);
+ dagActionStore.addDagAction(flowGroup, flowName, flowExecutionId_2, DagActionStore.DagActionType.RESUME);
dagManager = new MockedDagManager(ConfigUtils.propertiesToConfig(props));
dagManager.dagActionStore = Optional.of(dagActionStore);
dagManager.setActive(true);
@@ -129,9 +129,9 @@ public class DagManagerFlowTest {
// mock add spec
// for very first dag to be added, add dag action to store and check its deleted by the addDag call
- dagManager.getDagActionStore().get().addDagAction("group0", "flow0", Long.toString(flowExecutionId1), DagActionStore.FlowActionType.LAUNCH);
+ dagManager.getDagActionStore().get().addDagAction("group0", "flow0", Long.toString(flowExecutionId1), DagActionStore.DagActionType.LAUNCH);
dagManager.addDag(dag1, true, true);
- Assert.assertFalse(dagManager.getDagActionStore().get().exists("group0", "flow0", Long.toString(flowExecutionId1), DagActionStore.FlowActionType.LAUNCH));
+ Assert.assertFalse(dagManager.getDagActionStore().get().exists("group0", "flow0", Long.toString(flowExecutionId1), DagActionStore.DagActionType.LAUNCH));
dagManager.addDag(dag2, true, true);
dagManager.addDag(dag3, true, true);
@@ -338,7 +338,7 @@ public class DagManagerFlowTest {
String flowName = jobConfig.getString(ConfigurationKeys.FLOW_NAME_KEY);
// Add kill action to action store and call kill
- dagActionStore.addDagAction(flowGroup, flowName, String.valueOf(flowExecutionId), DagActionStore.FlowActionType.KILL);
+ dagActionStore.addDagAction(flowGroup, flowName, String.valueOf(flowExecutionId), DagActionStore.DagActionType.KILL);
dagManager.handleKillFlowRequest(flowGroup, flowName, flowExecutionId);
// Check that the kill dag action is removed
@@ -346,7 +346,7 @@ public class DagManagerFlowTest {
assertTrue(input -> {
try {
return !dagActionStore.exists(flowGroup, flowName, String.valueOf(flowExecutionId),
- DagActionStore.FlowActionType.KILL);
+ DagActionStore.DagActionType.KILL);
} catch (IOException | SQLException e) {
throw new RuntimeException(e);
}
@@ -355,13 +355,13 @@ public class DagManagerFlowTest {
// Add resume action to action store and call resume
- dagActionStore.addDagAction(flowGroup, flowName, String.valueOf(flowExecutionId), DagActionStore.FlowActionType.RESUME);
+ dagActionStore.addDagAction(flowGroup, flowName, String.valueOf(flowExecutionId), DagActionStore.DagActionType.RESUME);
dagManager.handleResumeFlowRequest(flowGroup, flowName, flowExecutionId);
// Check that the resume dag action is removed
AssertWithBackoff.create().maxSleepMs(5000).backoffFactor(1).assertTrue(input -> {
try {
- return !dagActionStore.exists(flowGroup, flowName, String.valueOf(flowExecutionId), DagActionStore.FlowActionType.RESUME);
+ return !dagActionStore.exists(flowGroup, flowName, String.valueOf(flowExecutionId), DagActionStore.DagActionType.RESUME);
} catch (IOException | SQLException e) {
throw new RuntimeException(e);
}
diff --git a/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngineTest.java b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngineTest.java
index 8c2882813..9d9f584a8 100644
--- a/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngineTest.java
+++ b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngineTest.java
@@ -80,14 +80,15 @@ public class DagProcessingEngineTest {
this.dagManagementStateStore = new MostlyMySqlDagManagementStateStore(config, null, null);
this.dagManagementStateStore.setTopologySpecMap(topologySpecMap);
this.dagManagementTaskStream =
- new DagManagementTaskStreamImpl(config, Optional.empty());
+ new DagManagementTaskStreamImpl(config, Optional.empty(), null);
this.dagProcFactory = new DagProcFactory(null);
DagProcessingEngine.DagProcEngineThread dagProcEngineThread =
new DagProcessingEngine.DagProcEngineThread(this.dagManagementTaskStream, this.dagProcFactory,
dagManagementStateStore);
this.dagTaskStream = spy(new MockedDagTaskStream());
DagProcessingEngine dagProcessingEngine =
- new DagProcessingEngine(config, dagTaskStream, this.dagProcFactory, dagManagementStateStore);
+ new DagProcessingEngine(config, Optional.ofNullable(dagTaskStream), Optional.ofNullable(this.dagProcFactory),
+ Optional.ofNullable(dagManagementStateStore));
}
static class MockedDagTaskStream implements DagTaskStream {
@@ -107,9 +108,9 @@ public class DagProcessingEngineTest {
throw new RuntimeException("Simulating an exception to stop the thread!");
}
if (i % FAILING_DAGS_FREQUENCY == 0 ) {
- return new MockedDagTask(new DagActionStore.DagAction("fg-" + i, "fn-" + i, "1234" + i, DagActionStore.FlowActionType.LAUNCH), true);
+ return new MockedDagTask(new DagActionStore.DagAction("fg-" + i, "fn-" + i, "1234" + i, "jn-" + i, DagActionStore.DagActionType.LAUNCH), true);
} else {
- return new MockedDagTask(new DagActionStore.DagAction("fg-" + i, "fn-" + i, "1234" + i, DagActionStore.FlowActionType.LAUNCH), false);
+ return new MockedDagTask(new DagActionStore.DagAction("fg-" + i, "fn-" + i, "1234" + i, "jn-" + i, DagActionStore.DagActionType.LAUNCH), false);
}
}
}
diff --git a/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/FlowTriggerHandlerTest.java b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/FlowLaunchHandlerTest.java
similarity index 87%
rename from gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/FlowTriggerHandlerTest.java
rename to gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/FlowLaunchHandlerTest.java
index 672892673..5a430191d 100644
--- a/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/FlowTriggerHandlerTest.java
+++ b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/FlowLaunchHandlerTest.java
@@ -27,16 +27,16 @@ import org.quartz.JobDataMap;
import org.testng.annotations.Test;
-public class FlowTriggerHandlerTest {
+public class FlowLaunchHandlerTest {
long eventToRevisit = 123000L;
long minimumLingerDurationMillis = 2000L;
- String cronExpression = FlowTriggerHandler.createCronFromDelayPeriod(minimumLingerDurationMillis);
+ String cronExpression = FlowLaunchHandler.createCronFromDelayPeriod(minimumLingerDurationMillis);
String cronExpressionSuffix = truncateFirstTwoFieldsOfCronExpression(cronExpression);
int schedulerBackOffMillis = 10;
- DagActionStore.DagAction flowAction = new DagActionStore.DagAction("flowName", "flowGroup",
- String.valueOf(eventToRevisit), DagActionStore.FlowActionType.LAUNCH);
+ DagActionStore.DagAction dagAction = new DagActionStore.DagAction("flowName", "flowGroup",
+ String.valueOf(eventToRevisit), "jobName", DagActionStore.DagActionType.LAUNCH);
MultiActiveLeaseArbiter.LeasedToAnotherStatus leasedToAnotherStatus =
- new MultiActiveLeaseArbiter.LeasedToAnotherStatus(flowAction, minimumLingerDurationMillis);
+ new MultiActiveLeaseArbiter.LeasedToAnotherStatus(dagAction, minimumLingerDurationMillis);
/**
* Remove first two fields from cron expression representing seconds and minutes to return truncated cron expression
@@ -64,7 +64,7 @@ public class FlowTriggerHandlerTest {
originalProperties.setProperty(ConfigurationKeys.SCHEDULER_PRESERVED_CONSENSUS_EVENT_TIME_MILLIS_KEY, "1");
oldJobDataMap.put(GobblinServiceJobScheduler.PROPERTIES_KEY, originalProperties);
- JobDataMap newJobDataMap = FlowTriggerHandler.updatePropsInJobDataMap(oldJobDataMap, leasedToAnotherStatus,
+ JobDataMap newJobDataMap = FlowLaunchHandler.updatePropsInJobDataMap(oldJobDataMap, leasedToAnotherStatus,
schedulerBackOffMillis);
Properties newProperties = (Properties) newJobDataMap.get(GobblinServiceJobScheduler.PROPERTIES_KEY);
Assert.assertTrue(newProperties.getProperty(ConfigurationKeys.JOB_SCHEDULE_KEY).endsWith(cronExpressionSuffix));
@@ -84,7 +84,7 @@ public class FlowTriggerHandlerTest {
Properties originalProperties = new Properties();
oldJobDataMap.put(GobblinServiceJobScheduler.PROPERTIES_KEY, originalProperties);
- JobDataMap newJobDataMap = FlowTriggerHandler.updatePropsInJobDataMap(oldJobDataMap, leasedToAnotherStatus,
+ JobDataMap newJobDataMap = FlowLaunchHandler.updatePropsInJobDataMap(oldJobDataMap, leasedToAnotherStatus,
schedulerBackOffMillis);
Properties newProperties = (Properties) newJobDataMap.get(GobblinServiceJobScheduler.PROPERTIES_KEY);
Assert.assertTrue(newProperties.getProperty(ConfigurationKeys.JOB_SCHEDULE_KEY).endsWith(cronExpressionSuffix));
@@ -99,7 +99,7 @@ public class FlowTriggerHandlerTest {
*/
@Test
public void testCreateSuffixForJobTrigger() {
- String suffix = FlowTriggerHandler.createSuffixForJobTrigger(leasedToAnotherStatus);
+ String suffix = FlowLaunchHandler.createSuffixForJobTrigger(leasedToAnotherStatus);
Assert.assertTrue(suffix.equals("reminder_for_" + eventToRevisit));
}
}
diff --git a/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/OrchestratorTest.java b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/OrchestratorTest.java
index 6a590dd26..8aacc6e0d 100644
--- a/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/OrchestratorTest.java
+++ b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/OrchestratorTest.java
@@ -110,7 +110,7 @@ public class OrchestratorTest {
this.serviceLauncher.addService(flowCatalog);
FlowStatusGenerator mockStatusGenerator = mock(FlowStatusGenerator.class);
- FlowTriggerHandler mockFlowTriggerHandler = mock(FlowTriggerHandler.class);
+ FlowLaunchHandler mockFlowTriggerHandler = mock(FlowLaunchHandler.class);
DagManager mockDagManager = mock(DagManager.class);
doNothing().when(mockDagManager).setTopologySpecMap(anyMap());
Config config = ConfigBuilder.create()
@@ -127,7 +127,7 @@ public class OrchestratorTest {
this.orchestrator = new Orchestrator(ConfigUtils.propertiesToConfig(orchestratorProperties),
this.topologyCatalog, mockDagManager, Optional.of(logger), mockStatusGenerator,
Optional.of(mockFlowTriggerHandler), new SharedFlowMetricsSingleton(ConfigUtils.propertiesToConfig(
- orchestratorProperties)), Optional.of(mock(FlowCatalog.class)), dagManagementStateStore, null);
+ orchestratorProperties)), Optional.of(mock(FlowCatalog.class)), Optional.of(dagManagementStateStore), null);
this.topologyCatalog.addListener(orchestrator);
this.flowCatalog.addListener(orchestrator);
// Start application
diff --git a/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/proc/LaunchDagProcTest.java b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/proc/LaunchDagProcTest.java
index d8ed2b96f..f00e35fbf 100644
--- a/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/proc/LaunchDagProcTest.java
+++ b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/proc/LaunchDagProcTest.java
@@ -78,7 +78,7 @@ public class LaunchDagProcTest {
FlowCompilationValidationHelper flowCompilationValidationHelper = mock(FlowCompilationValidationHelper.class);
doReturn(com.google.common.base.Optional.of(dag)).when(flowCompilationValidationHelper).createExecutionPlanIfValid(any());
LaunchDagProc launchDagProc = new LaunchDagProc(new LaunchDagTask(new DagActionStore.DagAction("fg", "fn",
- "12345", DagActionStore.FlowActionType.LAUNCH), null), flowCompilationValidationHelper);
+ "12345", "jn", DagActionStore.DagActionType.LAUNCH), null), flowCompilationValidationHelper);
launchDagProc.process(this.dagManagementStateStore);
int expectedNumOfSavingDagNodeStates = 1; // = number of start nodes
diff --git a/gobblin-service/src/test/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobSchedulerTest.java b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobSchedulerTest.java
index 06691ad36..c264bbfec 100644
--- a/gobblin-service/src/test/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobSchedulerTest.java
+++ b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobSchedulerTest.java
@@ -57,7 +57,7 @@ import org.apache.gobblin.service.modules.flow.MockedSpecCompiler;
import org.apache.gobblin.service.modules.flow.SpecCompiler;
import org.apache.gobblin.service.modules.flowgraph.Dag;
import org.apache.gobblin.service.modules.orchestration.AbstractUserQuotaManager;
-import org.apache.gobblin.service.modules.orchestration.FlowTriggerHandler;
+import org.apache.gobblin.service.modules.orchestration.FlowLaunchHandler;
import org.apache.gobblin.service.modules.orchestration.InMemoryUserQuotaManager;
import org.apache.gobblin.service.modules.orchestration.Orchestrator;
import org.apache.gobblin.service.modules.orchestration.UserQuotaManager;
@@ -356,7 +356,7 @@ public class GobblinServiceJobSchedulerTest {
GobblinServiceJobScheduler scheduler = new GobblinServiceJobScheduler("testscheduler",
ConfigFactory.empty(), Optional.absent(), Optional.of(flowCatalog), mockOrchestrator, schedulerService,
Optional.of(new InMemoryUserQuotaManager(quotaConfig)), Optional.absent(), false, Optional.of(Mockito.mock(
- FlowTriggerHandler.class)));
+ FlowLaunchHandler.class)));
schedulerService.startAsync().awaitRunning();
scheduler.startUp();
@@ -375,7 +375,7 @@ public class GobblinServiceJobSchedulerTest {
GobblinServiceJobScheduler schedulerWithWarmStandbyEnabled = new GobblinServiceJobScheduler("testscheduler",
ConfigFactory.empty(), Optional.absent(), Optional.of(flowCatalog), mockOrchestrator, schedulerService,
Optional.of(new InMemoryUserQuotaManager(quotaConfig)), Optional.absent(), true,
- Optional.of(Mockito.mock(FlowTriggerHandler.class)));
+ Optional.of(Mockito.mock(FlowLaunchHandler.class)));
schedulerWithWarmStandbyEnabled.startUp();
schedulerWithWarmStandbyEnabled.setActive(true);
@@ -398,7 +398,8 @@ public class GobblinServiceJobSchedulerTest {
Optional<FlowCatalog> flowCatalog, Orchestrator orchestrator, Optional<UserQuotaManager> quotaManager,
SchedulerService schedulerService, boolean isWarmStandbyEnabled) throws Exception {
super(serviceName, config, Optional.absent(), flowCatalog, orchestrator, schedulerService,
- quotaManager, Optional.absent(), isWarmStandbyEnabled, Optional.of(Mockito.mock(FlowTriggerHandler.class)));
+ quotaManager, Optional.absent(), isWarmStandbyEnabled, Optional.of(Mockito.mock(
+ FlowLaunchHandler.class)));
if (schedulerService != null) {
hasScheduler = true;
}