Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[GOBBLIN-1931] Refactor dag action updating method & add clarifying comment #3801

Merged
merged 4 commits into from
Oct 20, 2023
Merged

[GOBBLIN-1931] Refactor dag action updating method & add clarifying comment #3801

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
e2c0fd8
Refactor dag action updating method & add clarifying comment
Oct 18, 2023
531bf09
Log filtered out duplicate messages
Oct 19, 2023
d2a9b9b
logs and metrics for missing messages from change monitor
Oct 20, 2023
ec67462
Only add gobblin.service prefix for dagActionStoreChangeMonitor
Oct 20, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 3 additions & 4 deletions gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/DagActionStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,10 +49,9 @@ public FlowId getFlowId() {
/**
* Replace flow execution id with agreed upon event time to easily track the flow
*/
public static DagActionStore.DagAction updateFlowExecutionId(DagActionStore.DagAction flowAction,
long eventTimeMillis) {
return new DagActionStore.DagAction(flowAction.getFlowGroup(), flowAction.getFlowName(),
String.valueOf(eventTimeMillis), flowAction.getFlowActionType());
public DagAction updateFlowExecutionId(long eventTimeMillis) {
return new DagAction(this.getFlowGroup(), this.getFlowName(),
String.valueOf(eventTimeMillis), this.getFlowActionType());
}
}

Expand Down
6 changes: 3 additions & 3 deletions ...in-runtime/src/main/java/org/apache/gobblin/runtime/api/MysqlMultiActiveLeaseArbiter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -321,14 +321,14 @@ public LeaseAttemptStatus tryAcquireLease(DagActionStore.DagAction flowAction, l
// Lease is valid
if (leaseValidityStatus == 1) {
if (isWithinEpsilon) {
DagActionStore.DagAction updatedFlowAction = updateFlowExecutionId(flowAction, dbEventTimestamp.getTime());
DagActionStore.DagAction updatedFlowAction = flowAction.updateFlowExecutionId(dbEventTimestamp.getTime());
log.info("tryAcquireLease for [{}, is: {}, eventTimestamp: {}] - CASE 2: Same event, lease is valid",
updatedFlowAction, isReminderEvent ? "reminder" : "original", dbCurrentTimestamp.getTime());
// Utilize db timestamp for reminder
return new LeasedToAnotherStatus(updatedFlowAction,
dbLeaseAcquisitionTimestamp.getTime() + dbLinger - dbCurrentTimestamp.getTime());
}
DagActionStore.DagAction updatedFlowAction = updateFlowExecutionId(flowAction, dbCurrentTimestamp.getTime());
DagActionStore.DagAction updatedFlowAction = flowAction.updateFlowExecutionId(dbCurrentTimestamp.getTime());
log.info("tryAcquireLease for [{}, is: {}, eventTimestamp: {}] - CASE 3: Distinct event, lease is valid",
updatedFlowAction, isReminderEvent ? "reminder" : "original", dbCurrentTimestamp.getTime());
// Utilize db lease acquisition timestamp for wait time
Expand Down Expand Up @@ -518,7 +518,7 @@ protected LeaseAttemptStatus evaluateStatusAfterLeaseAttempt(int numRowsUpdated,
if (!selectInfoResult.getLeaseAcquisitionTimeMillis().isPresent()) {
return new NoLongerLeasingStatus();
}
DagActionStore.DagAction updatedFlowAction = updateFlowExecutionId(flowAction, selectInfoResult.eventTimeMillis);
DagActionStore.DagAction updatedFlowAction = flowAction.updateFlowExecutionId(selectInfoResult.eventTimeMillis);
if (numRowsUpdated == 1) {
log.info("Obtained lease for [{}, is: {}, eventTimestamp: {}] successfully!", updatedFlowAction,
isReminderEvent ? "reminder" : "original", selectInfoResult.eventTimeMillis);
Expand Down
4 changes: 2 additions & 2 deletions ...untime/src/test/java/org/apache/gobblin/runtime/api/MysqlMultiActiveLeaseArbiterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -217,7 +217,7 @@ public void testConditionallyAcquireLeaseIfFinishedLeasingStatement()
// Mark the resume action lease from above as completed by fabricating a LeaseObtainedStatus
MysqlMultiActiveLeaseArbiter.SelectInfoResult selectInfoResult =
mysqlMultiActiveLeaseArbiter.getRowInfo(resumeDagAction);
DagActionStore.DagAction updatedResumeDagAction = DagActionStore.DagAction.updateFlowExecutionId(resumeDagAction,
DagActionStore.DagAction updatedResumeDagAction = resumeDagAction.updateFlowExecutionId(
selectInfoResult.getEventTimeMillis());
boolean markedSuccess = mysqlMultiActiveLeaseArbiter.recordLeaseSuccess(new LeaseObtainedStatus(
updatedResumeDagAction, selectInfoResult.getLeaseAcquisitionTimeMillis().get()));
Expand Down Expand Up @@ -299,7 +299,7 @@ public void testReminderEventAcquireLeaseOnCompletedLease() throws IOException,
// Mark the resume action lease from above as completed by fabricating a LeaseObtainedStatus
MysqlMultiActiveLeaseArbiter.SelectInfoResult selectInfoResult =
mysqlMultiActiveLeaseArbiter.getRowInfo(resumeDagAction);
DagActionStore.DagAction updatedResumeDagAction = DagActionStore.DagAction.updateFlowExecutionId(resumeDagAction,
DagActionStore.DagAction updatedResumeDagAction = resumeDagAction.updateFlowExecutionId(
selectInfoResult.getEventTimeMillis());
boolean markedSuccess = mysqlMultiActiveLeaseArbiter.recordLeaseSuccess(new LeaseObtainedStatus(
updatedResumeDagAction, selectInfoResult.getLeaseAcquisitionTimeMillis().get()));
Expand Down
12 changes: 7 additions & 5 deletions ...ce/src/main/java/org/apache/gobblin/service/modules/orchestration/FlowTriggerHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,8 +112,12 @@ public void handleTriggerEvent(Properties jobProps, DagActionStore.DagAction flo
if (multiActiveLeaseArbiter.isPresent()) {
MultiActiveLeaseArbiter.LeaseAttemptStatus leaseAttemptStatus = multiActiveLeaseArbiter.get().tryAcquireLease(
flowAction, eventTimeMillis, isReminderEvent);
// 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) {
MultiActiveLeaseArbiter.LeaseObtainedStatus leaseObtainedStatus = (MultiActiveLeaseArbiter.LeaseObtainedStatus) leaseAttemptStatus;
MultiActiveLeaseArbiter.LeaseObtainedStatus leaseObtainedStatus = (MultiActiveLeaseArbiter.LeaseObtainedStatus)
leaseAttemptStatus;
this.leaseObtainedCount.inc();
if (persistFlowAction(leaseObtainedStatus)) {
log.info("Successfully persisted lease: [{}, eventTimestamp: {}] ", leaseObtainedStatus.getFlowAction(),
Expand All @@ -122,11 +126,9 @@ public void handleTriggerEvent(Properties jobProps, DagActionStore.DagAction flo
}
// If persisting the flow action failed, then we set another trigger for this event to occur immediately to
// re-attempt handling the event
DagActionStore.DagAction updatedFlowAction = DagActionStore.DagAction.updateFlowExecutionId(flowAction,
leaseObtainedStatus.getEventTimeMillis());
scheduleReminderForEvent(jobProps,
new MultiActiveLeaseArbiter.LeasedToAnotherStatus(updatedFlowAction, 0L),
eventTimeMillis);
new MultiActiveLeaseArbiter.LeasedToAnotherStatus(leaseObtainedStatus.getFlowAction(),
0L), eventTimeMillis);
return;
} else if (leaseAttemptStatus instanceof MultiActiveLeaseArbiter.LeasedToAnotherStatus) {
this.leasedToAnotherStatusCount.inc();
Expand Down
Loading