Skip to content

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

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 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 all commits
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: 3 additions & 1 deletion gobblin-runtime/src/main/java/org/apache/gobblin/runtime/kafka/HighLevelConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,7 @@ public abstract class HighLevelConsumer<K,V> extends AbstractIdleService {
*/
@Getter
private MetricContext metricContext;
private Counter messagesRead;
protected Counter messagesRead;
@Getter
private final GobblinKafkaConsumerClient gobblinKafkaConsumerClient;
private final ScheduledExecutorService consumerExecutor;
Expand Down Expand Up @@ -329,6 +329,8 @@ public void run() {
}
}
} catch (InterruptedException e) {
log.warn("Encountered exception while processing queue ", e);
// TODO: evaluate whether we should interrupt the thread or continue processing
Thread.currentThread().interrupt();
}
}
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
2 changes: 1 addition & 1 deletion gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/ChangeMonitorUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ public static boolean shouldProcessMessage(String changeIdentifier, LoadingCache
String operation, String timestamp) {
// If we've already processed a message with this timestamp and key before then skip duplicate message
if (cache.getIfPresent(changeIdentifier) != null) {
log.debug("Duplicate change event with identifier {}", changeIdentifier);
log.info("Duplicate change event with identifier {}", changeIdentifier);
return false;
}

Expand Down
3 changes: 2 additions & 1 deletion ...vice/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import org.apache.gobblin.kafka.client.DecodeableKafkaRecord;
import org.apache.gobblin.metrics.ContextAwareGauge;
import org.apache.gobblin.metrics.ContextAwareMeter;
import org.apache.gobblin.metrics.ServiceMetricNames;
import org.apache.gobblin.runtime.api.DagActionStore;
import org.apache.gobblin.runtime.api.FlowSpec;
import org.apache.gobblin.runtime.api.SpecNotFoundException;
Expand Down Expand Up @@ -216,7 +217,7 @@ protected void submitFlowToDagManagerHelper(String flowGroup, String flowName) {

@Override
protected void createMetrics() {
super.createMetrics();
super.messagesRead = this.getMetricContext().counter(ServiceMetricNames.GOBBLIN_SERVICE_PREFIX + RuntimeMetrics.GOBBLIN_KAFKA_HIGH_LEVEL_CONSUMER_MESSAGES_READ);
Copy link
Copy Markdown
Contributor

@Will-Lo Will-Lo Oct 20, 2023

Choose a reason for hiding this comment

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

If we're calling it here is there any different creating a counter vs a contextawarecounter?

Copy link
Copy Markdown
Contributor Author

@umustafi umustafi Oct 20, 2023

Choose a reason for hiding this comment

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

the super type is counter, so could not change to contextAwareCounter unfortunately.

this.killsInvoked = this.getMetricContext().contextAwareMeter(RuntimeMetrics.GOBBLIN_DAG_ACTION_STORE_MONITOR_KILLS_INVOKED);
this.resumesInvoked = this.getMetricContext().contextAwareMeter(RuntimeMetrics.GOBBLIN_DAG_ACTION_STORE_MONITOR_RESUMES_INVOKED);
this.flowsLaunched = this.getMetricContext().contextAwareMeter(RuntimeMetrics.GOBBLIN_DAG_ACTION_STORE_MONITOR_FLOWS_LAUNCHED);
Expand Down