Skip to content

[GOBBLIN-2016] Add eventTime to DagAction reminder key #3995

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

Merged
arjun4084346 merged 3 commits into from
Jul 9, 2024
Merged

[GOBBLIN-2016] Add eventTime to DagAction reminder key #3995

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
2fdf8c6
Add eventTime to DagAction reminder key
Jul 8, 2024
3747109
Clarify docstring
Jul 8, 2024
2d8042c
Rewrite javadoc to clearly describe
Jul 9, 2024
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
2 changes: 2 additions & 0 deletions ...rc/test/java/org/apache/gobblin/runtime/DagManagementDagActionStoreChangeMonitorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,10 +112,12 @@ public void testProcessMessageWithDelete() throws SchedulerException {
DagActionStore.DagAction dagAction = new DagActionStore.DagAction(FLOW_GROUP, FLOW_NAME, FLOW_EXECUTION_ID, JOB_NAME,
DagActionStore.DagActionType.ENFORCE_JOB_START_DEADLINE);
mockDagManagementDagActionStoreChangeMonitor.processMessageForTest(consumerRecord);
/* TODO: skip deadline removal for now and let them fire
verify(mockDagManagementDagActionStoreChangeMonitor.getDagActionReminderScheduler(), times(1))
.unscheduleReminderJob(eq(dagAction), eq(true));
verify(mockDagManagementDagActionStoreChangeMonitor.getDagActionReminderScheduler(), times(1))
.unscheduleReminderJob(eq(dagAction), eq(false));
*/
}

/**
Expand Down
42 changes: 27 additions & 15 deletions ...ain/java/org/apache/gobblin/service/modules/orchestration/DagActionReminderScheduler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,16 +80,16 @@ public void scheduleReminder(DagActionStore.LeaseParams leaseParams, long remind
boolean isDeadlineReminder)
throws SchedulerException {
JobDetail jobDetail = createReminderJobDetail(leaseParams, isDeadlineReminder);
Trigger trigger = createReminderJobTrigger(leaseParams.getDagAction(), reminderDurationMillis,
Trigger trigger = createReminderJobTrigger(leaseParams, reminderDurationMillis,
System::currentTimeMillis, isDeadlineReminder);
log.info("Reminder set for dagAction {} to fire after {} ms, isDeadlineTrigger: {}",
leaseParams.getDagAction(), reminderDurationMillis, isDeadlineReminder);
quartzScheduler.scheduleJob(jobDetail, trigger);
}

public void unscheduleReminderJob(DagActionStore.DagAction dagAction, boolean isDeadlineTrigger) throws SchedulerException {
log.info("Reminder unset for dagAction {}, isDeadlineTrigger: {}", dagAction, isDeadlineTrigger);
quartzScheduler.deleteJob(createJobKey(dagAction, isDeadlineTrigger));
public void unscheduleReminderJob(DagActionStore.LeaseParams leaseParams, boolean isDeadlineTrigger) throws SchedulerException {
log.info("Reminder unset for dagAction {}, isDeadlineTrigger: {}", leaseParams, isDeadlineTrigger);
quartzScheduler.deleteJob(createJobKey(leaseParams, isDeadlineTrigger));
}

/**
Expand Down Expand Up @@ -128,23 +128,35 @@ public void execute(JobExecutionContext context) {
}

/**
* Creates a key for the reminder job by concatenating all dagAction fields
* Creates a key for the reminder job by concatenating all dagAction fields and the eventTime of the dagAction.
*
* This ensures unique keys for multiple instances of the same action on the same flow execution that originate more
* than 'epsilon' apart. {@link MultiActiveLeaseArbiter} uses the eventTime to distinguish these distinct occurrences
* of the same action. This is necessary to prevent insertion failures due to previous reminders.
*
* Applicable only for KILL and RESUME actions; duplication for other actions is an error.
*/
public static String createDagActionReminderKey(DagActionStore.DagAction dagAction) {
return String.format("%s.%s.%s.%s.%s", dagAction.getFlowGroup(), dagAction.getFlowName(),
dagAction.getFlowExecutionId(), dagAction.getJobName(), dagAction.getDagActionType());
public static String createDagActionReminderKey(DagActionStore.LeaseParams leaseParams) {
DagActionStore.DagAction dagAction = leaseParams.getDagAction();
return String.join(".",
dagAction.getFlowGroup(),
dagAction.getFlowName(),
String.valueOf(dagAction.getFlowExecutionId()),
dagAction.getJobName(),
String.valueOf(dagAction.getDagActionType()),
String.valueOf(leaseParams.getEventTimeMillis()));
}

/**
* Creates a JobKey object for the reminder job where the name is the DagActionReminderKey from above and the group is
* the flowGroup
*/
public static JobKey createJobKey(DagActionStore.DagAction dagAction, boolean isDeadlineReminder) {
return new JobKey(createDagActionReminderKey(dagAction), isDeadlineReminder ? DeadlineReminderKeyGroup : RetryReminderKeyGroup);
public static JobKey createJobKey(DagActionStore.LeaseParams leaseParams, boolean isDeadlineReminder) {
return new JobKey(createDagActionReminderKey(leaseParams), isDeadlineReminder ? DeadlineReminderKeyGroup : RetryReminderKeyGroup);
}

private static TriggerKey createTriggerKey(DagActionStore.DagAction dagAction, boolean isDeadlineReminder) {
return new TriggerKey(createDagActionReminderKey(dagAction), isDeadlineReminder ? DeadlineReminderKeyGroup : RetryReminderKeyGroup);
private static TriggerKey createTriggerKey(DagActionStore.LeaseParams leaseParams, boolean isDeadlineReminder) {
return new TriggerKey(createDagActionReminderKey(leaseParams), isDeadlineReminder ? DeadlineReminderKeyGroup : RetryReminderKeyGroup);
}

/**
Expand All @@ -163,7 +175,7 @@ public static JobDetail createReminderJobDetail(DagActionStore.LeaseParams lease
dataMap.put(ReminderJob.FLOW_ACTION_EVENT_TIME_KEY, leaseParams.getEventTimeMillis());

return JobBuilder.newJob(ReminderJob.class)
.withIdentity(createJobKey(leaseParams.getDagAction(), isDeadlineReminder))
.withIdentity(createJobKey(leaseParams, isDeadlineReminder))
.usingJobData(dataMap)
.build();
}
Expand All @@ -173,10 +185,10 @@ public static JobDetail createReminderJobDetail(DagActionStore.LeaseParams lease
* with a job at any given time) that should fire after `reminderDurationMillis` millis. It uses
* `getCurrentTimeMillis` to determine the current time.
*/
public static Trigger createReminderJobTrigger(DagActionStore.DagAction dagAction, long reminderDurationMillis,
public static Trigger createReminderJobTrigger(DagActionStore.LeaseParams leaseParams, long reminderDurationMillis,
Supplier<Long> getCurrentTimeMillis, boolean isDeadlineReminder) {
return TriggerBuilder.newTrigger()
.withIdentity(createTriggerKey(dagAction, isDeadlineReminder))
.withIdentity(createTriggerKey(leaseParams, isDeadlineReminder))
.startAt(new Date(getCurrentTimeMillis.get() + reminderDurationMillis))
.build();
}
Expand Down
2 changes: 2 additions & 0 deletions .../java/org/apache/gobblin/service/monitoring/DagManagementDagActionStoreChangeMonitor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,12 +73,14 @@ protected void handleDagAction(String operation, DagActionStore.DagAction dagAct
break;
case "DELETE":
log.debug("Deleted dagAction from DagActionStore: {}", dagAction);
/* TODO: skip deadline removal for now and let them fire

@umustafi umustafi Jul 8, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

there's no event time in dagActionStore so cannot remove job from scheduler & this will be addressed in future pr to handle race condition between creating and deleting trigger

if (dagActionType == DagActionStore.DagActionType.ENFORCE_JOB_START_DEADLINE
|| dagActionType == DagActionStore.DagActionType.ENFORCE_FLOW_FINISH_DEADLINE) {
this.dagActionReminderScheduler.unscheduleReminderJob(dagAction, true);
// clear any deadline reminders as well as any retry reminders
this.dagActionReminderScheduler.unscheduleReminderJob(dagAction, false);
}
*/
break;
default:
log.warn(
Expand Down
41 changes: 35 additions & 6 deletions ...java/org/apache/gobblin/service/modules/orchestration/DagActionReminderSchedulerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,10 +23,13 @@

import org.quartz.JobDataMap;
import org.quartz.JobDetail;
import org.quartz.SchedulerException;
import org.quartz.Trigger;
import org.quartz.TriggerUtils;
import org.quartz.impl.StdSchedulerFactory;
import org.quartz.spi.OperableTrigger;
import org.testng.Assert;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

import com.google.common.base.Joiner;
Expand All @@ -39,31 +42,45 @@ public class DagActionReminderSchedulerTest {
String flowName = "fn";
long flowExecutionId = 123L;
String jobName = "jn";
long eventTimeMillis = 1234L;
long eventTimeMillis2 = 5678L;
String expectedKey = Joiner.on(".").join(flowGroup, flowName, flowExecutionId, jobName,
DagActionStore.DagActionType.LAUNCH.name());
DagActionStore.DagActionType.LAUNCH.name(), eventTimeMillis);
String expectedKey2 = Joiner.on(".").join(flowGroup, flowName, flowExecutionId, jobName,
DagActionStore.DagActionType.LAUNCH.name(), eventTimeMillis2);
DagActionStore.DagAction launchDagAction = new DagActionStore.DagAction(flowGroup, flowName, flowExecutionId, jobName,
DagActionStore.DagActionType.LAUNCH);
DagActionStore.LeaseParams launchLeaseParams = new DagActionStore.LeaseParams(launchDagAction, eventTimeMillis);
DagActionStore.LeaseParams launchLeaseParams2 = new DagActionStore.LeaseParams(launchDagAction, eventTimeMillis2);
DagActionReminderScheduler dagActionReminderScheduler;

@BeforeClass
private void setup() throws SchedulerException {
StdSchedulerFactory schedulerFactory = new StdSchedulerFactory();
schedulerFactory.getScheduler();
this.dagActionReminderScheduler = new DagActionReminderScheduler(schedulerFactory);
}

@Test
public void testCreateDagActionReminderKey() {
Assert.assertEquals(expectedKey, DagActionReminderScheduler.createDagActionReminderKey(launchDagAction));
Assert.assertEquals(expectedKey, DagActionReminderScheduler.createDagActionReminderKey(launchLeaseParams));
Assert.assertEquals(expectedKey2, DagActionReminderScheduler.createDagActionReminderKey(launchLeaseParams2));
}

@Test
public void testCreateReminderJobTrigger() {
long reminderDuration = 666L;
Supplier<Long> getCurrentTimeMillis = () -> 12345600000L;
Trigger reminderTrigger = DagActionReminderScheduler
.createReminderJobTrigger(launchDagAction, reminderDuration, getCurrentTimeMillis, false);
.createReminderJobTrigger(launchLeaseParams, reminderDuration, getCurrentTimeMillis, false);
Assert.assertEquals(reminderTrigger.getKey().toString(), DagActionReminderScheduler.RetryReminderKeyGroup + "." + expectedKey);
List<Date> fireTimes = TriggerUtils.computeFireTimes((OperableTrigger) reminderTrigger, null, 1);
Assert.assertEquals(fireTimes.get(0), new Date(reminderDuration + getCurrentTimeMillis.get()));
}

@Test
public void testCreateReminderJobDetail() {
long expectedEventTimeMillis = 55L;
JobDetail jobDetail = DagActionReminderScheduler.createReminderJobDetail(new DagActionStore.LeaseParams(launchDagAction, false, expectedEventTimeMillis), false);
JobDetail jobDetail = DagActionReminderScheduler.createReminderJobDetail(launchLeaseParams, false);
Assert.assertEquals(jobDetail.getKey().toString(), DagActionReminderScheduler.RetryReminderKeyGroup + "." + expectedKey);
JobDataMap dataMap = jobDetail.getJobDataMap();
Assert.assertEquals(dataMap.get(ConfigurationKeys.FLOW_GROUP_KEY), flowGroup);
Expand All @@ -72,6 +89,18 @@ public void testCreateReminderJobDetail() {
Assert.assertEquals(dataMap.get(ConfigurationKeys.JOB_NAME_KEY), jobName);
Assert.assertEquals(dataMap.get(DagActionReminderScheduler.ReminderJob.FLOW_ACTION_TYPE_KEY),
DagActionStore.DagActionType.LAUNCH);
Assert.assertEquals(dataMap.get(DagActionReminderScheduler.ReminderJob.FLOW_ACTION_EVENT_TIME_KEY), expectedEventTimeMillis);
Assert.assertEquals(dataMap.get(DagActionReminderScheduler.ReminderJob.FLOW_ACTION_EVENT_TIME_KEY), launchLeaseParams.getEventTimeMillis());
}

/*
Add deadline reminders for multiple launches of the same flow and assert no exception is thrown and they can be
deleted as well.
*/
@Test
public void testRemindersForMultipleFlowExecutions() throws SchedulerException {
this.dagActionReminderScheduler.scheduleReminder(launchLeaseParams, 50000, true);
this.dagActionReminderScheduler.scheduleReminder(launchLeaseParams2, 50000, true);
this.dagActionReminderScheduler.unscheduleReminderJob(launchLeaseParams, true);
this.dagActionReminderScheduler.unscheduleReminderJob(launchLeaseParams2, true);
}
}