phet commented on code in PR #3899:
URL: https://github.com/apache/gobblin/pull/3899#discussion_r1543632884
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/core/GobblinServiceManager.java:
##########
@@ -251,10 +252,24 @@ public static GobblinServiceManager create(String
serviceName, String serviceId,
}
public static GobblinServiceManager create(GobblinServiceConfiguration
serviceConfiguration) {
- GobblinServiceGuiceModule guiceModule = new
GobblinServiceGuiceModule(serviceConfiguration);
+ GOBBLIN_SERVICE_GUICE_MODULE = new
GobblinServiceGuiceModule(serviceConfiguration);
Review Comment:
are we sure `create` would be called at most once? if not, is it OK to
replace this ref again and again?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementTaskStreamImpl.java:
##########
@@ -51,20 +70,33 @@ public class DagManagementTaskStreamImpl implements
DagManagement, DagTaskStream
@Inject(optional=true)
protected Optional<DagActionStore> dagActionStore;
- protected Optional<ReminderSettingDagProcLeaseArbiter>
reminderSettingDagProcLeaseArbiter;
+ protected MultiActiveLeaseArbiter dagActionProcessingLeaseArbiter;
+ protected Optional<DagActionReminderScheduler> dagActionReminderScheduler;
+ private final boolean isMultiActiveExecutionEnabled;
@Inject
private static final int MAX_HOUSEKEEPING_THREAD_DELAY = 180;
private final BlockingQueue<DagActionStore.DagAction> dagActionQueue = new
LinkedBlockingQueue<>();
+ private final String MISSING_OPTIONAL_ERROR_MESSAGE =
String.format("Multi-active execution enabled but required "
+ + "instance %s is absent.",
DagActionReminderScheduler.class.getSimpleName());
- // TODO: need to pass reference to DagProcLeaseArbiter without creating a
circular reference in Guice
@Inject
public DagManagementTaskStreamImpl(Config config, Optional<DagActionStore>
dagActionStore,
- Optional<ReminderSettingDagProcLeaseArbiter>
reminderSettingDagProcLeaseArbiter) {
+ @Named(ConfigurationKeys.PROCESSING_LEASE_ARBITER_NAME)
MultiActiveLeaseArbiter dagActionProcessingLeaseArbiter,
+ Optional<DagActionReminderScheduler> dagActionReminderScheduler,
+ @Named(InjectionNames.MULTI_ACTIVE_EXECUTION_ENABLED) boolean
isMultiActiveExecutionEnabled) {
this.config = config;
+ if (!dagActionStore.isPresent()) {
+ throw new RuntimeException("DagProcessingEngine should not be enabled
without dagActionStore enabled.");
+ }
this.dagActionStore = dagActionStore;
- this.reminderSettingDagProcLeaseArbiter =
reminderSettingDagProcLeaseArbiter;
+ this.dagActionProcessingLeaseArbiter = dagActionProcessingLeaseArbiter;
+ this.dagActionReminderScheduler = dagActionReminderScheduler;
+ this.isMultiActiveExecutionEnabled = isMultiActiveExecutionEnabled;
MetricContext metricContext =
Instrumented.getMetricContext(ConfigUtils.configToState(ConfigFactory.empty()),
getClass());
this.eventSubmitter = new EventSubmitter.Builder(metricContext,
"org.apache.gobblin.service").build();
+ if (this.isMultiActiveExecutionEnabled &&
!this.dagActionReminderScheduler.isPresent()) {
Review Comment:
don't we *always* want reminder scheduling?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementTaskStreamImpl.java:
##########
@@ -51,20 +70,33 @@ public class DagManagementTaskStreamImpl implements
DagManagement, DagTaskStream
@Inject(optional=true)
protected Optional<DagActionStore> dagActionStore;
- protected Optional<ReminderSettingDagProcLeaseArbiter>
reminderSettingDagProcLeaseArbiter;
+ protected MultiActiveLeaseArbiter dagActionProcessingLeaseArbiter;
+ protected Optional<DagActionReminderScheduler> dagActionReminderScheduler;
+ private final boolean isMultiActiveExecutionEnabled;
@Inject
private static final int MAX_HOUSEKEEPING_THREAD_DELAY = 180;
private final BlockingQueue<DagActionStore.DagAction> dagActionQueue = new
LinkedBlockingQueue<>();
+ private final String MISSING_OPTIONAL_ERROR_MESSAGE =
String.format("Multi-active execution enabled but required "
+ + "instance %s is absent.",
DagActionReminderScheduler.class.getSimpleName());
- // TODO: need to pass reference to DagProcLeaseArbiter without creating a
circular reference in Guice
@Inject
public DagManagementTaskStreamImpl(Config config, Optional<DagActionStore>
dagActionStore,
- Optional<ReminderSettingDagProcLeaseArbiter>
reminderSettingDagProcLeaseArbiter) {
+ @Named(ConfigurationKeys.PROCESSING_LEASE_ARBITER_NAME)
MultiActiveLeaseArbiter dagActionProcessingLeaseArbiter,
+ Optional<DagActionReminderScheduler> dagActionReminderScheduler,
+ @Named(InjectionNames.MULTI_ACTIVE_EXECUTION_ENABLED) boolean
isMultiActiveExecutionEnabled) {
this.config = config;
+ if (!dagActionStore.isPresent()) {
+ throw new RuntimeException("DagProcessingEngine should not be enabled
without dagActionStore enabled.");
+ }
Review Comment:
I suggested elsewhere to have this be non-`Optional`. if that's not
possible or desired, let's add a comment here explaining why, given it's
strange to declare an `Optional` that isn't actually allowed to be
##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagManagerFlowTest.java:
##########
Review Comment:
NBD, but strange name for a test (at least I'm unaware of a `DagManagerFlow`
class). if you understand why, please just add class-level javadoc to inform
maintainers
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementTaskStreamImpl.java:
##########
@@ -51,20 +70,33 @@ public class DagManagementTaskStreamImpl implements
DagManagement, DagTaskStream
@Inject(optional=true)
protected Optional<DagActionStore> dagActionStore;
- protected Optional<ReminderSettingDagProcLeaseArbiter>
reminderSettingDagProcLeaseArbiter;
+ protected MultiActiveLeaseArbiter dagActionProcessingLeaseArbiter;
+ protected Optional<DagActionReminderScheduler> dagActionReminderScheduler;
+ private final boolean isMultiActiveExecutionEnabled;
@Inject
private static final int MAX_HOUSEKEEPING_THREAD_DELAY = 180;
private final BlockingQueue<DagActionStore.DagAction> dagActionQueue = new
LinkedBlockingQueue<>();
+ private final String MISSING_OPTIONAL_ERROR_MESSAGE =
String.format("Multi-active execution enabled but required "
+ + "instance %s is absent.",
DagActionReminderScheduler.class.getSimpleName());
- // TODO: need to pass reference to DagProcLeaseArbiter without creating a
circular reference in Guice
@Inject
public DagManagementTaskStreamImpl(Config config, Optional<DagActionStore>
dagActionStore,
- Optional<ReminderSettingDagProcLeaseArbiter>
reminderSettingDagProcLeaseArbiter) {
+ @Named(ConfigurationKeys.PROCESSING_LEASE_ARBITER_NAME)
MultiActiveLeaseArbiter dagActionProcessingLeaseArbiter,
+ Optional<DagActionReminderScheduler> dagActionReminderScheduler,
+ @Named(InjectionNames.MULTI_ACTIVE_EXECUTION_ENABLED) boolean
isMultiActiveExecutionEnabled) {
this.config = config;
+ if (!dagActionStore.isPresent()) {
+ throw new RuntimeException("DagProcessingEngine should not be enabled
without dagActionStore enabled.");
+ }
this.dagActionStore = dagActionStore;
- this.reminderSettingDagProcLeaseArbiter =
reminderSettingDagProcLeaseArbiter;
+ this.dagActionProcessingLeaseArbiter = dagActionProcessingLeaseArbiter;
+ this.dagActionReminderScheduler = dagActionReminderScheduler;
+ this.isMultiActiveExecutionEnabled = isMultiActiveExecutionEnabled;
MetricContext metricContext =
Instrumented.getMetricContext(ConfigUtils.configToState(ConfigFactory.empty()),
getClass());
this.eventSubmitter = new EventSubmitter.Builder(metricContext,
"org.apache.gobblin.service").build();
+ if (this.isMultiActiveExecutionEnabled &&
!this.dagActionReminderScheduler.isPresent()) {
+ throw new RuntimeException(MISSING_OPTIONAL_ERROR_MESSAGE);
Review Comment:
I don't actually see it as clearer for the error message to be separately
initialized in a `static`. isn't it used only in this single place? what's
the motivation?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/DagTask.java:
##########
@@ -38,26 +39,31 @@
*/
@Alpha
+@Slf4j
public abstract class DagTask {
@Getter public final DagActionStore.DagAction dagAction;
private final LeaseAttemptStatus.LeaseObtainedStatus leaseObtainedStatus;
+ private final DagActionStore dagActionStore;
@Getter protected final DagManager.DagId dagId;
- public DagTask(DagActionStore.DagAction dagAction,
LeaseAttemptStatus.LeaseObtainedStatus leaseObtainedStatus) {
+ public DagTask(DagActionStore.DagAction dagAction,
LeaseAttemptStatus.LeaseObtainedStatus leaseObtainedStatus,
+ DagActionStore dagActionStore) {
this.dagAction = dagAction;
this.leaseObtainedStatus = leaseObtainedStatus;
+ this.dagActionStore = dagActionStore;
this.dagId = DagManagerUtils.generateDagId(dagAction.getFlowGroup(),
dagAction.getFlowName(), dagAction.getFlowExecutionId());
}
public abstract <T> T host(DagTaskVisitor<T> visitor);
/**
- * Any cleanup work, e.g. releasing lease if it was acquired earlier, may be
done in this method.
+ * Any cleanup work, including removing the dagAction from the
dagActionStore and completing the lease acquired to
+ * work on this task, is done in this method.
* Returns true if concluding dag task finished successfully otherwise false.
Review Comment:
we deliberated on which order to delete vs. conclude lease. please capture
our reasoning here (as well as what could potentially result if failure between
first and second steps
nit: use javadoc `@return` syntax
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/LaunchDagProc.java:
##########
@@ -58,28 +56,27 @@
* An implementation for {@link DagProc} that launches a new job.
*/
@Slf4j
-@RequiredArgsConstructor
public class LaunchDagProc extends DagProc<Optional<Dag<JobExecutionPlan>>,
Optional<Dag<JobExecutionPlan>>> {
- private final LaunchDagTask launchDagTask;
private final FlowCompilationValidationHelper
flowCompilationValidationHelper;
// todo - this is not orchestration delay and should be renamed. keeping it
the same because DagManager is also using
// the same name
private static final AtomicLong orchestrationDelayCounter = new
AtomicLong(0);
+
+ public LaunchDagProc(LaunchDagTask launchDagTask,
FlowCompilationValidationHelper flowCompilationValidationHelper) {
+ super(launchDagTask);
+ this.flowCompilationValidationHelper = flowCompilationValidationHelper;
+ }
+
static {
metricContext.register(
metricContext.newContextAwareGauge(ServiceMetricNames.FLOW_ORCHESTRATION_DELAY,
orchestrationDelayCounter::get));
}
- @Override
- protected DagManager.DagId getDagId() {
- return this.launchDagTask.getDagId();
- }
-
@Override
protected Optional<Dag<JobExecutionPlan>> initialize(DagManagementStateStore
dagManagementStateStore)
throws IOException {
try {
- DagActionStore.DagAction dagAction = this.launchDagTask.getDagAction();
+ DagActionStore.DagAction dagAction = this.getDagTask().getDagAction();
Review Comment:
since I don't see the `DagProc` derived classes actually using the `DagTask`
anywhere, it might be possible to be `private`.
then provide a `protected DagProc DagProc::getDagAction()` for these derived
classes
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementTaskStreamImpl.java:
##########
@@ -84,19 +116,12 @@ public boolean hasNext() {
@Override
public DagTask next() {
- /*TODO: this requires use of lease arbiter, in single-active execution
lease arbiter will not be present and we can
- provide a dummy LeaseObtainedStatus or create alternate route
- */
- if (!this.reminderSettingDagProcLeaseArbiter.isPresent()) {
- throw new RuntimeException("DagManagement not initialized in
multi-active execution mode when required.");
- }
try {
LeaseAttemptStatus leaseAttemptStatus = null;
DagActionStore.DagAction dagAction = null;
while (!(leaseAttemptStatus instanceof
LeaseAttemptStatus.LeaseObtainedStatus)) {
dagAction = this.dagActionQueue.take(); //`take` blocks till element
is not available
- // TODO: need to handle reminder events and flag them
- leaseAttemptStatus =
this.reminderSettingDagProcLeaseArbiter.get().tryAcquireLease(dagAction,
System.currentTimeMillis(), false, false);
+ leaseAttemptStatus = retrieveLeaseStatus(dagAction);
}
return createDagTask(dagAction, (LeaseAttemptStatus.LeaseObtainedStatus)
leaseAttemptStatus);
Review Comment:
nit: to me this reads more clearly:
```
while (true) {
DagActionStore.DagAction dagAction = this.dagActionQueue.take(); //
blocks until element available
LeaseAttemptStatus leaseAttemptStatus = retrieveLeaseStatus(dagAction);
if (leaseAttemptStatus instanceof
LeaseAttemptStatus.LeaseObtainedStatus) {
return createDagTask(dagAction,
(LeaseAttemptStatus.LeaseObtainedStatus) leaseAttemptStatus);
}
}
```
##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagManagementTaskStreamImplTest.java:
##########
@@ -74,8 +75,11 @@ public void setUp() throws Exception {
topologySpecMap.put(specExecURI, topologySpec);
MostlyMySqlDagManagementStateStore dagManagementStateStore = new
MostlyMySqlDagManagementStateStore(config, null, null, null);
dagManagementStateStore.setTopologySpecMap(topologySpecMap);
+ // TODO: create tests for cases with multiActiveExecutionEnabled
Review Comment:
same Q here: what specifically should we test in that case?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementTaskStreamImpl.java:
##########
@@ -106,14 +131,47 @@ public DagTask next() {
return null;
Review Comment:
probably best to put the `try / catch` within the `while` loop. that way no
occasionally returning `null`.
also, should we really catch `Throwable`? seems extreme. maybe just
`Exception`?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementTaskStreamImpl.java:
##########
@@ -51,20 +70,33 @@ public class DagManagementTaskStreamImpl implements
DagManagement, DagTaskStream
@Inject(optional=true)
protected Optional<DagActionStore> dagActionStore;
- protected Optional<ReminderSettingDagProcLeaseArbiter>
reminderSettingDagProcLeaseArbiter;
+ protected MultiActiveLeaseArbiter dagActionProcessingLeaseArbiter;
+ protected Optional<DagActionReminderScheduler> dagActionReminderScheduler;
+ private final boolean isMultiActiveExecutionEnabled;
@Inject
private static final int MAX_HOUSEKEEPING_THREAD_DELAY = 180;
private final BlockingQueue<DagActionStore.DagAction> dagActionQueue = new
LinkedBlockingQueue<>();
+ private final String MISSING_OPTIONAL_ERROR_MESSAGE =
String.format("Multi-active execution enabled but required "
+ + "instance %s is absent.",
DagActionReminderScheduler.class.getSimpleName());
- // TODO: need to pass reference to DagProcLeaseArbiter without creating a
circular reference in Guice
@Inject
public DagManagementTaskStreamImpl(Config config, Optional<DagActionStore>
dagActionStore,
- Optional<ReminderSettingDagProcLeaseArbiter>
reminderSettingDagProcLeaseArbiter) {
+ @Named(ConfigurationKeys.PROCESSING_LEASE_ARBITER_NAME)
MultiActiveLeaseArbiter dagActionProcessingLeaseArbiter,
+ Optional<DagActionReminderScheduler> dagActionReminderScheduler,
+ @Named(InjectionNames.MULTI_ACTIVE_EXECUTION_ENABLED) boolean
isMultiActiveExecutionEnabled) {
Review Comment:
why do we need to know (at this low level) whether MA execution is enabled?
##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagProcessingEngineTest.java:
##########
@@ -78,8 +80,10 @@ public void setUp() throws Exception {
topologySpecMap.put(specExecURI, topologySpec);
this.dagManagementStateStore = new
MostlyMySqlDagManagementStateStore(config, null, null, null);
this.dagManagementStateStore.setTopologySpecMap(topologySpecMap);
+ // TODO create another test with multiActiveExecution enabled
Review Comment:
is that flag doing anything? what should we test?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MultiActiveLeaseArbiterFactory.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.service.modules.orchestration;
+
+import java.io.IOException;
+import java.util.Objects;
+
+import com.typesafe.config.Config;
+
+import javax.inject.Provider;
+import lombok.extern.slf4j.Slf4j;
+
+
+/**
+ * An abstract base class for {@link MultiActiveLeaseArbiter} factories that
use a specific configuration key.
+ * Subclasses must provide a key to use in the constructor.
+ */
+@Slf4j
+public abstract class MultiActiveLeaseArbiterFactory implements
Provider<MultiActiveLeaseArbiter> {
+ private final Config leaseArbiterConfig;
+ private final String configPrefix;
+
+ public MultiActiveLeaseArbiterFactory(Config config, String configPrefix) {
+ Objects.requireNonNull(config);
+ this.configPrefix = Objects.requireNonNull(configPrefix);
+ if (!config.hasPath(configPrefix)) {
+ throw new RuntimeException(String.format("Unable to initialize
multiActiveLeaseArbiter due to missing "
+ + "configurations that should be prefixed by %s.", configPrefix));
+ }
+ this.leaseArbiterConfig = config.getConfig(configPrefix);
+ log.info("Lease arbiter will be initialized with config {}",
leaseArbiterConfig);
+ }
+
+ @Override
+ public MultiActiveLeaseArbiter get() {
+ try {
+ return new InstrumentedLeaseArbiter(this.leaseArbiterConfig, new
MysqlMultiActiveLeaseArbiter(leaseArbiterConfig), configPrefix);
Review Comment:
nit: separate each ctor invocation on its own line
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/LaunchDagProc.java:
##########
@@ -192,6 +189,7 @@ private void submitJobToExecutor(DagManagementStateStore
dagManagementStateStore
}
}
+ @Override
Review Comment:
good to add. since I believe we agreed to remove elsewhere, `@Override`
would catch mistakes from merging
--
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]
