phet commented on code in PR #4031:
URL: https://github.com/apache/gobblin/pull/4031#discussion_r1745965920
##########
gobblin-api/src/main/java/org/apache/gobblin/configuration/ConfigurationKeys.java:
##########
@@ -1071,13 +1071,17 @@ public class ConfigurationKeys {
* Configuration properties related to Flows
*/
public static final String FLOW_RUN_IMMEDIATELY = "flow.runImmediately";
- public static final String GOBBLIN_FLOW_SLA_TIME = "gobblin.flow.sla.time";
- public static final String GOBBLIN_FLOW_SLA_TIME_UNIT =
"gobblin.flow.sla.timeunit";
- public static final String DEFAULT_GOBBLIN_FLOW_SLA_TIME_UNIT =
TimeUnit.MINUTES.name();
- public static final String GOBBLIN_JOB_START_SLA_TIME =
"gobblin.job.start.sla.time";
- public static final String GOBBLIN_JOB_START_SLA_TIME_UNIT =
"gobblin.job.start.sla.timeunit";
- public static final long FALLBACK_GOBBLIN_JOB_START_SLA_TIME = 10L;
- public static final String FALLBACK_GOBBLIN_JOB_START_SLA_TIME_UNIT =
TimeUnit.MINUTES.name();
+ /*
+ The following config names are different from variable name to maintain
backward compatibility.
Review Comment:
nit: can you name this as a "TODO", so anyone searching for string finds it?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java:
##########
@@ -407,20 +288,12 @@ protected static class LaunchSubmissionMetricProxy {
public LaunchSubmissionMetricProxy() {}
- public void markSuccess() {
- getSuccessMeter().mark();
- }
-
Review Comment:
we need to keep both of these `markSuccess` methods. It appears to be an
oversight that `DagManagementDagActionStoreChangeMonitor` is not marking the
success metric upon success, the way L370 (above in
`submitFlowToDagManagerHelper` is).
I believe it belongs here, just before the `break`):
```
case REEVALUATE :
case RESUME:
dagManagement.addDagAction(new
DagActionStore.LeaseParams(dagAction));
break;
default:
```
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java:
##########
@@ -252,123 +234,14 @@ protected void
processMessage(DecodeableKafkaRecord<String, DagActionStoreChange
dagActionsSeenCache.put(changeIdentifier, changeIdentifier);
}
- protected void handleDagAction(String operation, DagActionStore.DagAction
dagAction, String flowGroup,
- String flowName, long flowExecutionId, DagActionStore.DagActionType
dagActionType) {
- // We only expect INSERT and DELETE operations done to this table. INSERTs
correspond to any type of
- // {@link DagActionStore.FlowActionType} flow requests that have to be
processed. DELETEs require no action.
- try {
- switch (operation) {
- case "INSERT":
- handleDagAction(dagAction, false);
- this.dagProcEngineMetrics.markDagActionsObserved(dagActionType);
- break;
- case "UPDATE":
- // TODO: change this warning message and process updates if for
launch or reevaluate type
- log.warn("Received an UPDATE action to the DagActionStore when
values in this store are never supposed to be "
- + "updated. Flow group: {} name {} executionId {} were
updated to action {}", flowGroup, flowName,
- flowExecutionId, dagActionType);
- this.unexpectedErrors.mark();
- break;
- case "DELETE":
- log.debug("Deleted dagAction from DagActionStore: {}", dagAction);
- break;
- default:
- log.warn(
- "Received unsupported change type of operation {}. Expected
values to be in [INSERT, UPDATE, DELETE]",
- operation);
- this.unexpectedErrors.mark();
- break;
- }
- } catch (Exception e) {
- log.warn("Ran into unexpected error processing DagActionStore changes:
", e);
- this.unexpectedErrors.mark();
- }
- }
Review Comment:
shouldn't we leave all of this impl as-is for the derived class to continue
using?
that said, I do agree w/ making **the other form** of `handleDagAction`
`abstract`
--
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]
