phet commented on code in PR #4078:
URL: https://github.com/apache/gobblin/pull/4078#discussion_r1866494762
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/workflow/impl/ProcessWorkUnitsWorkflowImpl.java:
##########
@@ -72,9 +72,20 @@ private CommitStats performWork(WUProcessingSpec workSpec) {
searchAttributes =
TemporalWorkFlowUtils.generateGaasSearchAttributes(jobState.getProperties());
NestingExecWorkflow<WorkUnitClaimCheck> processingWorkflow =
createProcessingWorkflow(workSpec, searchAttributes);
- int workunitsProcessed =
- processingWorkflow.performWorkload(WorkflowAddr.ROOT, workload, 0,
workSpec.getTuning().getMaxBranchesPerTree(),
- workSpec.getTuning().getMaxSubTreesPerTree(), Optional.empty());
+
+ int workunitsProcessed = 0;
+ try {
+ workunitsProcessed =
processingWorkflow.performWorkload(WorkflowAddr.ROOT, workload, 0,
+ workSpec.getTuning().getMaxBranchesPerTree(),
workSpec.getTuning().getMaxSubTreesPerTree(), Optional.empty());
+ } catch (Exception e) {
+ log.error("Exception occurred in performing workload,proceeding with
commit step", e);
+ return proceedWithCommitStepAndReturnCommitStats(workSpec,
searchAttributes, workunitsProcessed);
+ }
+ return proceedWithCommitStepAndReturnCommitStats(workSpec,
searchAttributes, workunitsProcessed);
+ }
+
+ private CommitStats
proceedWithCommitStepAndReturnCommitStats(WUProcessingSpec workSpec,
+ Map<String, Object> searchAttributes, int workunitsProcessed) {
if (workunitsProcessed > 0) {
Review Comment:
won't this always be zero, and hence never run, in cases where
`processingWorkflow.performWorkload` fails?
tip: perhaps initialize `Optional<Integer> workunitsProcessed =
Optional.empty()`, to discern cases where it's unknown vs. known to be zero.
only inhibit commit when actually known to be zero
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/workflow/impl/ProcessWorkUnitsWorkflowImpl.java:
##########
@@ -72,9 +72,20 @@ private CommitStats performWork(WUProcessingSpec workSpec) {
searchAttributes =
TemporalWorkFlowUtils.generateGaasSearchAttributes(jobState.getProperties());
NestingExecWorkflow<WorkUnitClaimCheck> processingWorkflow =
createProcessingWorkflow(workSpec, searchAttributes);
- int workunitsProcessed =
- processingWorkflow.performWorkload(WorkflowAddr.ROOT, workload, 0,
workSpec.getTuning().getMaxBranchesPerTree(),
- workSpec.getTuning().getMaxSubTreesPerTree(), Optional.empty());
+
+ int workunitsProcessed = 0;
+ try {
+ workunitsProcessed =
processingWorkflow.performWorkload(WorkflowAddr.ROOT, workload, 0,
+ workSpec.getTuning().getMaxBranchesPerTree(),
workSpec.getTuning().getMaxSubTreesPerTree(), Optional.empty());
+ } catch (Exception e) {
+ log.error("Exception occurred in performing workload,proceeding with
commit step", e);
+ return proceedWithCommitStepAndReturnCommitStats(workSpec,
searchAttributes, workunitsProcessed);
Review Comment:
we decided on the semantics where the job itself would fail when only
partially successful, but this swallows the exception, which may allow it to
continue on to overall success.
--
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]
