Will-Lo commented on code in PR #3790:
URL: https://github.com/apache/gobblin/pull/3790#discussion_r1343240323
##########
gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/MysqlMultiActiveLeaseArbiter.java:
##########
@@ -210,8 +242,23 @@ public LeaseAttemptStatus
tryAcquireLease(DagActionStore.DagAction flowAction, l
int dbLinger = getResult.get().getDbLinger();
Timestamp dbCurrentTimestamp = getResult.get().getDbCurrentTimestamp();
+ // For reminder event, we can stop early if the reminder eventTimeMillis
is older than the current event in the db
+ // because db laundering tells us that the currently worked on db event
is newer and will have its own reminders
+ if (isReminderEvent) {
+ if (eventTimeMillis < dbEventTimestamp.getTime()) {
+ log.info("tryAcquireLease for [{}, eventTimestamp: {}] -
dbEventTimeMillis: {} - A new event trigger is being"
+ + " worked on, so this older reminder will be dropped.",
flowAction, eventTimeMillis, dbEventTimestamp);
+ return new NoLongerLeasingStatus();
+ } if (eventTimeMillis > dbEventTimestamp.getTime()) {
Review Comment:
small nit, should put this if statement on a new line given it is not
related to the old one
##########
gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/MysqlMultiActiveLeaseArbiter.java:
##########
@@ -86,16 +88,26 @@ protected interface CheckedFunction<T, R> {
private final int epsilon;
private final int linger;
private String thisTableGetInfoStatement;
+ private String thisTableGetInfoStatementForReminder;
private String thisTableSelectAfterInsertStatement;
private String thisTableAcquireLeaseIfMatchingAllStatement;
private String thisTableAcquireLeaseIfFinishedStatement;
// TODO: define retention on this table
+ /*
+ Notes:
+ - Set `event_timestamp` default value to turn off timestamp auto-updates
for row modifications which alters this col
+ in an unexpected way upon completing the lease
+ - Upon reading any timestamps from MySQL we convert the timezone from
session (default) to UTC to consistently
+ use epoch-millis in UTC locally
+ - Upon using any timestamps from local we convert the timezone from UTC to
session
+ - We desire millisecond level precision and denote that with `(3)` for the
TIMESTAMP types
+ */
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, "
- + "event_timestamp TIMESTAMP, "
- + "lease_acquisition_timestamp TIMESTAMP NULL DEFAULT CURRENT_TIMESTAMP,
"
+ + "event_timestamp TIMESTAMP(3) DEFAULT CURRENT_TIMESTAMP(3), "
Review Comment:
I thought we were going to make this an int?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
