arjun4084346 commented on code in PR #3896:
URL: https://github.com/apache/gobblin/pull/3896#discussion_r1550238052
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/KafkaJobStatusMonitor.java:
##########
@@ -193,7 +197,18 @@ protected void
processMessage(DecodeableKafkaRecord<byte[],byte[]> message) {
org.apache.gobblin.configuration.State jobStatus =
parseJobStatus(gobblinTrackingEvent);
if (jobStatus != null) {
try (Timer.Context context =
getMetricContext().timer(GET_AND_SET_JOB_STATUS).time()) {
- addJobStatusToStateStore(jobStatus, this.stateStore,
this.eventProducer);
+ Optional<org.apache.gobblin.configuration.State> updatedJobStatus
= addJobStatusToStateStore(jobStatus, this.stateStore);
+ boolean isJobStatusUpdated = updatedJobStatus.isPresent();
+ // todo - retried/resumed jobs *may* not be handled here, we may
want to create their dag action elsewhere
+ if (isJobStatusUpdated) {
+ jobStatus = updatedJobStatus.get();
+ this.eventProducer.emitObservabilityEvent(jobStatus);
+ String flowName =
jobStatus.getProp(TimingEvent.FlowEventConstants.FLOW_NAME_FIELD);
+ String flowGroup =
jobStatus.getProp(TimingEvent.FlowEventConstants.FLOW_GROUP_FIELD);
+ String flowExecutionId =
jobStatus.getProp(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD);
+ String jobName =
jobStatus.getProp(TimingEvent.FlowEventConstants.JOB_NAME_FIELD);
+ this.dagActionStore.addJobDagAction(flowGroup, flowName,
flowExecutionId, jobName, DagActionStore.DagActionType.REEVALUATE);
Review Comment:
This suggestion will be useful when DagActionStore starts taking DagAction
as a param in addFlowDagAction, but it is not doing that right now.
Also, only one or two callers of addFlowDagAction use state to derive fg/fn.
One caller uses FlowId to get them
--
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]
