[
https://issues.apache.org/jira/browse/GOBBLIN-2175?focusedWorklogId=946385&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-946385
]
ASF GitHub Bot logged work on GOBBLIN-2175:
-------------------------------------------
Author: ASF GitHub Bot
Created on: 02/Dec/24 19:37
Start Date: 02/Dec/24 19:37
Worklog Time Spent: 10m
Work Description: 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.
Issue Time Tracking
-------------------
Worklog Id: (was: 946385)
Time Spent: 20m (was: 10m)
> Fix partial commit in temporal flow
> -----------------------------------
>
> Key: GOBBLIN-2175
> URL: https://issues.apache.org/jira/browse/GOBBLIN-2175
> Project: Apache Gobblin
> Issue Type: Bug
> Reporter: Aditya Pratap Singh
> Priority: Major
> Time Spent: 20m
> Remaining Estimate: 0h
>
> Fix partial commit in temporal flow
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
