pratapaditya04 commented on code in PR #4078:
URL: https://github.com/apache/gobblin/pull/4078#discussion_r1870954475
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/workflow/impl/ProcessWorkUnitsWorkflowImpl.java:
##########
@@ -72,10 +73,33 @@ 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());
- if (workunitsProcessed > 0) {
+
+ Optional<Integer> workunitsProcessed = Optional.empty();
+ try {
+ workunitsProcessed =
Optional.of(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);
+ // We want to mark the GaaS flow as failure, in case performWorkFlow
fails, but we still want to go ahead with commiting the workunits which were
processed before failure
+ sendFailureEventToGaaS(workSpec);
+ return proceedWithCommitStepAndReturnCommitStats(workSpec,
searchAttributes, workunitsProcessed);
+ }
+ return proceedWithCommitStepAndReturnCommitStats(workSpec,
searchAttributes, workunitsProcessed);
+ }
+
+ private void sendFailureEventToGaaS(WUProcessingSpec workSpec) {
Review Comment:
addressed
--
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]
