Skip to content

[GOBBLIN-1851] Unit tests for MysqlMultiActiveLeaseArbiter with Single Participant #3715

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
Will-Lo merged 10 commits into from
Jul 18, 2023
Merged

[GOBBLIN-1851] Unit tests for MysqlMultiActiveLeaseArbiter with Single Participant #3715

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
9b7f505
Unit tests and corresponding fixes for 6 lease acquisition cases
Jul 10, 2023
48134fb
remove logs
Jul 12, 2023
1feddee
address review comments with small changes
Jul 13, 2023
8b8de03
clean up naming, logs, java doc
Jul 17, 2023
84c2941
close resultSets
Jul 17, 2023
fdbf7a8
check if test passes on github
Jul 17, 2023
7c7bb2c
fix unit test
Jul 18, 2023
b43a587
insert or update constants into table correctly
Jul 18, 2023
a048d84
rename prop
Jul 18, 2023
b82d45a
address comments
Jul 18, 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
125 changes: 80 additions & 45 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 @@ -87,8 +87,8 @@ protected interface CheckedFunction<T, R> {
private final String constantsTableName;
private final int epsilon;
private final int linger;
String formattedGetInfoStatement;
String formattedSelectAfterInsertStatement;
private String thisTableGetInfoStatement;
private String thisTableSelectAfterInsertStatement;

// TODO: define retention on this table
private static final String CREATE_LEASE_ARBITER_TABLE_STATEMENT = "CREATE TABLE IF NOT EXISTS %s ("
Expand All @@ -113,10 +113,10 @@ protected interface CheckedFunction<T, R> {
// event_timestamp), leaseValidityStatus (1 if lease has not expired, 2 if expired, 3 if column is NULL or no longer
// leasing)
protected static final String GET_EVENT_INFO_STATEMENT = "SELECT event_timestamp, lease_acquisition_timestamp, "
+ "TIMESTAMPDIFF(microsecond, event_timestamp, CURRENT_TIMESTAMP) / 1000 <= epsilon as isWithinEpsilon, CASE "
+ "TIMESTAMPDIFF(microsecond, event_timestamp, CURRENT_TIMESTAMP) / 1000 <= epsilon as is_within_epsilon, CASE "
+ "WHEN CURRENT_TIMESTAMP < DATE_ADD(lease_acquisition_timestamp, INTERVAL linger*1000 MICROSECOND) then 1 "
+ "WHEN CURRENT_TIMESTAMP >= DATE_ADD(lease_acquisition_timestamp, INTERVAL linger*1000 MICROSECOND) then 2 "
+ "ELSE 3 END as leaseValidityStatus, linger, CURRENT_TIMESTAMP FROM %s, %s " + WHERE_CLAUSE_TO_MATCH_KEY;
+ "ELSE 3 END as lease_validity_status, linger, CURRENT_TIMESTAMP FROM %s, %s " + 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 CONDITIONALLY_ACQUIRE_LEASE_IF_NEW_ROW_STATEMENT = "INSERT INTO %s (flow_group, "
Expand Down Expand Up @@ -149,9 +149,9 @@ public MysqlMultiActiveLeaseArbiter(Config config) throws IOException {
ConfigurationKeys.DEFAULT_SCHEDULER_EVENT_EPSILON_MILLIS);
this.linger = ConfigUtils.getInt(config, ConfigurationKeys.SCHEDULER_EVENT_LINGER_MILLIS_KEY,
ConfigurationKeys.DEFAULT_SCHEDULER_EVENT_LINGER_MILLIS);
this.formattedGetInfoStatement = String.format(GET_EVENT_INFO_STATEMENT, this.leaseArbiterTableName,
this.thisTableGetInfoStatement = String.format(GET_EVENT_INFO_STATEMENT, this.leaseArbiterTableName,
this.constantsTableName);
this.formattedSelectAfterInsertStatement = String.format(SELECT_AFTER_INSERT_STATEMENT, this.leaseArbiterTableName,
this.thisTableSelectAfterInsertStatement = String.format(SELECT_AFTER_INSERT_STATEMENT, this.leaseArbiterTableName,
this.constantsTableName);
this.dataSource = MysqlDataSourceFactory.get(config, SharedResourcesBrokerFactory.getImplicitBroker());
String createArbiterStatement = String.format(
Expand All @@ -175,10 +175,16 @@ private void initializeConstantsTable() throws IOException {

Optional<Integer> count = withPreparedStatement(String.format(GET_ROW_COUNT_STATEMENT, this.constantsTableName), getStatement -> {
ResultSet resultSet = getStatement.executeQuery();
if (resultSet.next()) {
return Optional.of(resultSet.getInt(1));
try {
if (resultSet.next()) {
return Optional.of(resultSet.getInt(1));
}
return Optional.absent();
} finally {
if (resultSet != null) {
resultSet.close();
}
}
return Optional.absent();
}, true);

// Only insert epsilon and linger values from config if this table does not contain pre-existing values.
Expand All @@ -197,24 +203,30 @@ private void initializeConstantsTable() throws IOException {
public LeaseAttemptStatus tryAcquireLease(DagActionStore.DagAction flowAction, long eventTimeMillis)
throws IOException {
// Check table for an existing entry for this flow action and event time
Optional<GetEventInfoResult> getResult = withPreparedStatement(
formattedGetInfoStatement,
Optional<GetEventInfoResult> getResult = withPreparedStatement(thisTableGetInfoStatement,
getInfoStatement -> {
int i = 0;
getInfoStatement.setString(++i, flowAction.getFlowGroup());
getInfoStatement.setString(++i, flowAction.getFlowName());
getInfoStatement.setString(++i, flowAction.getFlowExecutionId());
getInfoStatement.setString(++i, flowAction.getFlowActionType().toString());
ResultSet resultSet = getInfoStatement.executeQuery();
if (!resultSet.next()) {
return Optional.absent();
try {
if (!resultSet.next()) {
return Optional.absent();
}
return Optional.of(createGetInfoResult(resultSet));
} finally {
if (resultSet != null) {
resultSet.close();
}
}
return createGetInfoResult(resultSet);
}, true);

try {
if (!getResult.isPresent()) {
log.debug("CASE 1: no existing row for this flow action, then go ahead and insert");
log.debug("tryAcquireLease for [{}, eventTimestamp: {}] - CASE 1: no existing row for this flow action, then go"
+ " ahead and insert", flowAction, eventTimeMillis);
String formattedAcquireLeaseNewRowStatement =
String.format(CONDITIONALLY_ACQUIRE_LEASE_IF_NEW_ROW_STATEMENT, this.leaseArbiterTableName);
int numRowsUpdated = withPreparedStatement(formattedAcquireLeaseNewRowStatement,
Expand All @@ -233,24 +245,27 @@ public LeaseAttemptStatus tryAcquireLease(DagActionStore.DagAction flowAction, l
int dbLinger = getResult.get().getDbLinger();
Comment thread
umustafi marked this conversation as resolved.
Timestamp dbCurrentTimestamp = getResult.get().getDbCurrentTimestamp();

log.info("Multi-active arbiter replacing local trigger event timestamp with database one {}: "
+ "[{}, triggerEventTimestamp: {}]", dbCurrentTimestamp, flowAction, eventTimeMillis);
log.info("Multi-active arbiter replacing local trigger event timestamp [{}, triggerEventTimestamp: {}] with "
+ "database eventTimestamp {}", flowAction, eventTimeMillis, dbCurrentTimestamp.getTime());

// Lease is valid
if (leaseValidityStatus == 1) {
if (isWithinEpsilon) {
log.debug("CASE 2: Same event, lease is valid");
log.debug("tryAcquireLease for [{}, eventTimestamp: {}] - CASE 2: Same event, lease is valid", flowAction,
dbCurrentTimestamp.getTime());
// Utilize db timestamp for reminder
return new LeasedToAnotherStatus(flowAction, dbEventTimestamp.getTime(),
dbLeaseAcquisitionTimestamp.getTime() + dbLinger - System.currentTimeMillis());
}
log.debug("CASE 3: Distinct event, lease is valid");
log.debug("tryAcquireLease for [{}, eventTimestamp: {}] - CASE 3: Distinct event, lease is valid", flowAction,
dbCurrentTimestamp.getTime());
// Utilize db lease acquisition timestamp for wait time
return new LeasedToAnotherStatus(flowAction, dbCurrentTimestamp.getTime(),
dbLeaseAcquisitionTimestamp.getTime() + dbLinger - System.currentTimeMillis());
}
else if (leaseValidityStatus == 2) {
log.debug("CASE 4: Lease is out of date (regardless of whether same or distinct event)");
log.debug("tryAcquireLease for [{}, eventTimestamp: {}] - CASE 4: Lease is out of date (regardless of whether "
+ "same or distinct event)", flowAction, dbCurrentTimestamp.getTime());
if (isWithinEpsilon) {
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,
Expand All @@ -268,10 +283,12 @@ else if (leaseValidityStatus == 2) {
return evaluateStatusAfterLeaseAttempt(numRowsUpdated, flowAction);
} // No longer leasing this event
if (isWithinEpsilon) {
log.debug("CASE 5: Same event, no longer leasing event in db: terminate");
log.debug("tryAcquireLease for [{}, eventTimestamp: {}] - CASE 5: Same event, no longer leasing event in db: "
+ "terminate", flowAction, dbCurrentTimestamp.getTime());
return new NoLongerLeasingStatus();
}
log.debug("CASE 6: Distinct event, no longer leasing event in db");
log.debug("tryAcquireLease for [{}, eventTimestamp: {}] - CASE 6: Distinct event, no longer leasing event in "
+ "db", flowAction, dbCurrentTimestamp.getTime());
// Use our event to acquire lease, check for previous db eventTimestamp and NULL leaseAcquisitionTimestamp
String formattedAcquireLeaseIfFinishedStatement =
String.format(CONDITIONALLY_ACQUIRE_LEASE_IF_FINISHED_LEASING_STATEMENT, this.leaseArbiterTableName);
Expand All @@ -287,32 +304,43 @@ else if (leaseValidityStatus == 2) {
}
}

protected Optional<GetEventInfoResult> createGetInfoResult(ResultSet resultSet) {
protected GetEventInfoResult createGetInfoResult(ResultSet resultSet) throws SQLException {
try {
// Extract values from result set
Timestamp dbEventTimestamp = resultSet.getTimestamp("event_timestamp");
Timestamp dbLeaseAcquisitionTimestamp = resultSet.getTimestamp("lease_acquisition_timestamp");
boolean withinEpsilon = resultSet.getBoolean("isWithinEpsilon");
int leaseValidityStatus = resultSet.getInt("leaseValidityStatus");
boolean withinEpsilon = resultSet.getBoolean("is_within_epsilon");
int leaseValidityStatus = resultSet.getInt("lease_validity_status");
int dbLinger = resultSet.getInt("linger");
Timestamp dbCurrentTimestamp = resultSet.getTimestamp("CURRENT_TIMESTAMP");
return Optional.of(new GetEventInfoResult(dbEventTimestamp, dbLeaseAcquisitionTimestamp, withinEpsilon, leaseValidityStatus,
dbLinger, dbCurrentTimestamp));
} catch (SQLException exception) {
log.warn("Failed to retrieve values from GET event info query resultSet. Exception: ", exception);
// Note: this will proceed to CASE 1 of acquiring a lease above
return Optional.absent();
return new GetEventInfoResult(dbEventTimestamp, dbLeaseAcquisitionTimestamp, withinEpsilon, leaseValidityStatus,
dbLinger, dbCurrentTimestamp);
} catch (SQLException e) {
throw e;

@phet phet Jul 17, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

  1. if this is all you want, it's the implicit behavior, which you need not write explicitly.

  2. even so, elsewhere, we wrap SQLException in an IOException. do we want that here too... or is there already a higher layer wrapping around this invocation where that will happen for us?

@umustafi umustafi Jul 17, 2023

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.

I meant to wrap in IOException as we do that in other places, updating to wrap it.

} finally {
if (resultSet != null) {
resultSet.close();
}
}
}

protected SelectInfoResult createSelectInfoResult(ResultSet resultSet) throws SQLException {
if (!resultSet.next()) {
log.error("Expected num rows and lease_acquisition_timestamp returned from query but received nothing");
try {
if (!resultSet.next()) {
resultSet.close();
Comment thread
umustafi marked this conversation as resolved.
Outdated
log.error("Expected num rows and lease_acquisition_timestamp returned from query but received nothing");

@Will-Lo Will-Lo Jul 18, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I'm confused here, so if there's no item after in the result set we log and error but still try to parse the current result set results?

@umustafi umustafi Jul 18, 2023 •
edited
Loading

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.

Oh good catch, I want this code path to terminate so instead will through an IO Error.

}
long eventTimeMillis = resultSet.getTimestamp(1).getTime();
long leaseAcquisitionTimeMillis = resultSet.getTimestamp(2).getTime();
int dbLinger = resultSet.getInt(3);
return new SelectInfoResult(eventTimeMillis, leaseAcquisitionTimeMillis, dbLinger);
} catch (SQLException e) {
throw e;
} finally {
if (resultSet != null) {
resultSet.close();
}
}
long eventTimeMillis = resultSet.getTimestamp(1).getTime();
long leaseAcquisitionTimeMillis = resultSet.getTimestamp(2).getTime();
int dbLinger = resultSet.getInt(3);
return new SelectInfoResult(eventTimeMillis, leaseAcquisitionTimeMillis, dbLinger);
}

/**
Expand All @@ -326,13 +354,20 @@ protected LeaseAttemptStatus evaluateStatusAfterLeaseAttempt(int numRowsUpdated,
DagActionStore.DagAction flowAction)
throws SQLException, IOException {
// Fetch values in row after attempted insert
SelectInfoResult selectInfoResult = withPreparedStatement(formattedSelectAfterInsertStatement,
SelectInfoResult selectInfoResult = withPreparedStatement(thisTableSelectAfterInsertStatement,
selectStatement -> {
completeWhereClauseMatchingKeyPreparedStatement(selectStatement, flowAction);
return createSelectInfoResult(selectStatement.executeQuery());
ResultSet resultSet = selectStatement.executeQuery();
try {
return createSelectInfoResult(resultSet);
} finally {
if (resultSet != null) {
resultSet.close();
}
}
}, true);
if (numRowsUpdated == 1) {
log.debug("Obtained lease for flowAction {} at eventTime {} successfully!", flowAction,
log.debug("Obtained lease for [{}, eventTimestamp: {}] successfully!", flowAction,
selectInfoResult.eventTimeMillis);
return new LeaseObtainedStatus(flowAction, selectInfoResult.eventTimeMillis,
selectInfoResult.getLeaseAcquisitionTimeMillis());
Expand Down Expand Up @@ -446,6 +481,7 @@ protected <T> T withPreparedStatement(String sql, CheckedFunction<PreparedStatem
if (shouldCommit) {
connection.commit();
}
statement.close();
return result;
} catch (SQLException e) {
log.warn("Received SQL exception that can result from invalid connection. Checking if validation query is set {} "
Expand All @@ -456,11 +492,10 @@ protected <T> T withPreparedStatement(String sql, CheckedFunction<PreparedStatem


/**
Class used to store information from initial SELECT query resultSet to be used for understanding the state of the
flow action event's lease in the arbiter store and act accordingly.
* DTO for arbiter's current lease state for a FlowActionEvent
*/
@Data
class GetEventInfoResult {
static class GetEventInfoResult {
private Timestamp dbEventTimestamp;
private Timestamp dbLeaseAcquisitionTimestamp;
private boolean withinEpsilon;
Expand All @@ -481,10 +516,10 @@ class GetEventInfoResult {
}

/**
Class used to store information from SELECT query used to determine status of lease acquisition attempt.
DTO for result of SELECT query used to determine status of lease acquisition attempt
*/
@Data
class SelectInfoResult {
static class SelectInfoResult {
private long eventTimeMillis;
private long leaseAcquisitionTimeMillis;
private int dbLinger;
Expand Down
17 changes: 17 additions & 0 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
@@ -1,3 +1,20 @@
/*
* 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.runtime.api;

import com.typesafe.config.Config;
Expand Down