beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From boyu...@apache.org
Subject [beam] branch master updated: Release bundle processor when any exceptions during processing.
Date Mon, 21 Dec 2020 19:45:58 GMT
This is an automated email from the ASF dual-hosted git repository.

boyuanz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new e9fa621  Release bundle processor when any exceptions during processing.
     new 060b083  Merge pull request #13568 from [BEAM-3245] Release bundle processor when
any exceptions during processing.
e9fa621 is described below

commit e9fa621a10ce6a84179fba1d54b413535a1a46a7
Author: Boyuan Zhang <boyuanz@google.com>
AuthorDate: Wed Dec 16 10:44:57 2020 -0800

    Release bundle processor when any exceptions during processing.
---
 .../fn/harness/control/ProcessBundleHandler.java   | 87 ++++++++++++----------
 1 file changed, 46 insertions(+), 41 deletions(-)

diff --git a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/ProcessBundleHandler.java
b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/ProcessBundleHandler.java
index e93c67b..4ecb5f5 100644
--- a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/ProcessBundleHandler.java
+++ b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/ProcessBundleHandler.java
@@ -293,55 +293,60 @@ public class ProcessBundleHandler {
     ExecutionStateTracker stateTracker = bundleProcessor.getStateTracker();
     QueueingBeamFnDataClient queueingClient = bundleProcessor.getQueueingClient();
 
-    try (HandleStateCallsForBundle beamFnStateClient = bundleProcessor.getBeamFnStateClient())
{
-      try (Closeable closeTracker = stateTracker.activate()) {
-        // Already in reverse topological order so we don't need to do anything.
-        for (ThrowingRunnable startFunction : startFunctionRegistry.getFunctions()) {
-          LOG.debug("Starting function {}", startFunction);
-          startFunction.run();
-        }
+    try {
+      try (HandleStateCallsForBundle beamFnStateClient = bundleProcessor.getBeamFnStateClient())
{
+        try (Closeable closeTracker = stateTracker.activate()) {
+          // Already in reverse topological order so we don't need to do anything.
+          for (ThrowingRunnable startFunction : startFunctionRegistry.getFunctions()) {
+            LOG.debug("Starting function {}", startFunction);
+            startFunction.run();
+          }
 
-        queueingClient.drainAndBlock();
+          queueingClient.drainAndBlock();
 
-        // Need to reverse this since we want to call finish in topological order.
-        for (ThrowingRunnable finishFunction :
-            Lists.reverse(finishFunctionRegistry.getFunctions())) {
-          LOG.debug("Finishing function {}", finishFunction);
-          finishFunction.run();
+          // Need to reverse this since we want to call finish in topological order.
+          for (ThrowingRunnable finishFunction :
+              Lists.reverse(finishFunctionRegistry.getFunctions())) {
+            LOG.debug("Finishing function {}", finishFunction);
+            finishFunction.run();
+          }
         }
-      }
 
-      // Add all checkpointed residuals to the response.
-      response.addAllResidualRoots(bundleProcessor.getSplitListener().getResidualRoots());
-
-      // TODO(BEAM-6597): This should be reporting monitoring infos using the short id system.
-      // Get start bundle Execution Time Metrics.
-      response.addAllMonitoringInfos(
-          bundleProcessor.getStartFunctionRegistry().getExecutionTimeMonitoringInfos());
-      // Get process bundle Execution Time Metrics.
-      response.addAllMonitoringInfos(
-          bundleProcessor.getpCollectionConsumerRegistry().getExecutionTimeMonitoringInfos());
-      // Get finish bundle Execution Time Metrics.
-      response.addAllMonitoringInfos(
-          bundleProcessor.getFinishFunctionRegistry().getExecutionTimeMonitoringInfos());
-      // Extract MonitoringInfos that come from the metrics container registry.
-      response.addAllMonitoringInfos(
-          bundleProcessor.getMetricsContainerRegistry().getMonitoringInfos());
-      // Add any additional monitoring infos that the "runners" report explicitly.
-      for (ProgressRequestCallback progressRequestCallback :
-          bundleProcessor.getProgressRequestCallbacks()) {
-        response.addAllMonitoringInfos(progressRequestCallback.getMonitoringInfos());
-      }
+        // Add all checkpointed residuals to the response.
+        response.addAllResidualRoots(bundleProcessor.getSplitListener().getResidualRoots());
+
+        // TODO(BEAM-6597): This should be reporting monitoring infos using the short id
system.
+        // Get start bundle Execution Time Metrics.
+        response.addAllMonitoringInfos(
+            bundleProcessor.getStartFunctionRegistry().getExecutionTimeMonitoringInfos());
+        // Get process bundle Execution Time Metrics.
+        response.addAllMonitoringInfos(
+            bundleProcessor.getpCollectionConsumerRegistry().getExecutionTimeMonitoringInfos());
+        // Get finish bundle Execution Time Metrics.
+        response.addAllMonitoringInfos(
+            bundleProcessor.getFinishFunctionRegistry().getExecutionTimeMonitoringInfos());
+        // Extract MonitoringInfos that come from the metrics container registry.
+        response.addAllMonitoringInfos(
+            bundleProcessor.getMetricsContainerRegistry().getMonitoringInfos());
+        // Add any additional monitoring infos that the "runners" report explicitly.
+        for (ProgressRequestCallback progressRequestCallback :
+            bundleProcessor.getProgressRequestCallbacks()) {
+          response.addAllMonitoringInfos(progressRequestCallback.getMonitoringInfos());
+        }
 
-      if (!bundleProcessor.getBundleFinalizationCallbackRegistrations().isEmpty()) {
-        finalizeBundleHandler.registerCallbacks(
-            bundleProcessor.getInstructionId(),
-            ImmutableList.copyOf(bundleProcessor.getBundleFinalizationCallbackRegistrations()));
-        response.setRequiresFinalization(true);
+        if (!bundleProcessor.getBundleFinalizationCallbackRegistrations().isEmpty()) {
+          finalizeBundleHandler.registerCallbacks(
+              bundleProcessor.getInstructionId(),
+              ImmutableList.copyOf(bundleProcessor.getBundleFinalizationCallbackRegistrations()));
+          response.setRequiresFinalization(true);
+        }
       }
-
       bundleProcessorCache.release(
           request.getProcessBundle().getProcessBundleDescriptorId(), bundleProcessor);
+    } catch (Exception e) {
+      bundleProcessorCache.release(
+          request.getProcessBundle().getProcessBundleDescriptorId(), bundleProcessor);
+      throw e;
     }
     return BeamFnApi.InstructionResponse.newBuilder().setProcessBundle(response);
   }


Mime
View raw message