Blazer-007 commented on code in PR #4093:
URL: https://github.com/apache/gobblin/pull/4093#discussion_r1938791672
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/workflow/impl/NestingExecOfProcessWorkUnitWorkflowImpl.java:
##########
@@ -17,39 +17,26 @@
package org.apache.gobblin.temporal.ddm.workflow.impl;
-import io.temporal.activity.ActivityOptions;
-import io.temporal.common.RetryOptions;
+import java.util.Properties;
+
import io.temporal.workflow.Async;
import io.temporal.workflow.Promise;
import io.temporal.workflow.Workflow;
-import java.time.Duration;
+import org.apache.gobblin.temporal.ddm.activity.ActivityType;
import org.apache.gobblin.temporal.ddm.activity.ProcessWorkUnit;
+import org.apache.gobblin.temporal.ddm.util.TemporalActivityUtils;
import org.apache.gobblin.temporal.ddm.work.WorkUnitClaimCheck;
import
org.apache.gobblin.temporal.util.nesting.workflow.AbstractNestingExecWorkflowImpl;
/** {@link
org.apache.gobblin.temporal.util.nesting.workflow.NestingExecWorkflow} for
{@link ProcessWorkUnit} */
public class NestingExecOfProcessWorkUnitWorkflowImpl extends
AbstractNestingExecWorkflowImpl<WorkUnitClaimCheck, Integer> {
- public static final Duration processWorkUnitStartToCloseTimeout =
Duration.ofHours(3); // TODO: make configurable... also add activity heartbeats
-
- // RetryOptions specify how to automatically handle retries when Activities
fail.
- private static final RetryOptions ACTIVITY_RETRY_OPTS =
RetryOptions.newBuilder()
- .setInitialInterval(Duration.ofSeconds(3))
- .setMaximumInterval(Duration.ofSeconds(100))
- .setBackoffCoefficient(2)
- .setMaximumAttempts(4)
- .build();
-
- private static final ActivityOptions ACTIVITY_OPTS =
ActivityOptions.newBuilder()
- .setStartToCloseTimeout(processWorkUnitStartToCloseTimeout)
- .setRetryOptions(ACTIVITY_RETRY_OPTS)
- .build();
-
- private final ProcessWorkUnit activityStub =
Workflow.newActivityStub(ProcessWorkUnit.class, ACTIVITY_OPTS);
@Override
- protected Promise<Integer> launchAsyncActivity(final WorkUnitClaimCheck wu) {
- return Async.function(activityStub::processWorkUnit, wu);
+ protected Promise<Integer> launchAsyncActivity(final WorkUnitClaimCheck wu,
final Properties props) {
Review Comment:
Unit tests for older temporal classes needs to be done separately as it will
required dedicated effort for this which is beyond the scope of this PR
--
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: dev-unsubscr...@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
