dcache-bulk: fix handling of uncaught exception during perform

Backport for 9.0 and 9.1 of https://rb.dcache.org/r/14162/ Acked-by: Lea
dCache · Nov 17, 2023 · d1dc7b5 · d1dc7b5
1 parent f6cd9f5
commit d1dc7b5
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 9 deletions.
diff --git a/...s/dcache-bulk/src/main/java/org/dcache/services/bulk/job/AbstractRequestContainerJob.java b/...s/dcache-bulk/src/main/java/org/dcache/services/bulk/job/AbstractRequestContainerJob.java
@@ -62,7 +62,6 @@ LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 import static org.dcache.services.bulk.activity.BulkActivity.MINIMALLY_REQUIRED_ATTRIBUTES;
 import static org.dcache.services.bulk.util.BulkRequestTarget.State.CANCELLED;
 import static org.dcache.services.bulk.util.BulkRequestTarget.State.COMPLETED;
-import static org.dcache.services.bulk.util.BulkRequestTarget.State.FAILED;
 import static org.dcache.services.bulk.util.BulkRequestTarget.State.READY;
 import static org.dcache.services.bulk.util.BulkRequestTarget.State.SKIPPED;
 import static org.dcache.services.bulk.util.BulkRequestTarget.computeFsPath;
@@ -342,7 +341,7 @@ public void uncaughtException(Thread t, Throwable e) {
          */
         containerState = ContainerState.STOP;
         target.setErrorObject(e);
-        update(FAILED);
+        update();
         ThreadGroup group = t.getThreadGroup();
         if (group != null) {
             group.uncaughtException(t, e);
@@ -353,13 +352,7 @@ public void uncaughtException(Thread t, Throwable e) {
 
     public void update(State state) {
         if (target.setState(state)) {
-            try {
-                targetStore.update(target.getId(), target.getState(), target.getErrorType(),
-                      target.getErrorMessage());
-            } catch (BulkStorageException e) {
-                LOGGER.error("{}, updateJobState: {}", ruid, e.toString());
-            }
-            signalStateChange();
+            update();
         }
     }
 
@@ -547,4 +540,14 @@ private synchronized void setRunThread(Thread runThread) {
             this.runThread.setUncaughtExceptionHandler(this);
         }
     }
+
+    private void update() {
+        try {
+            targetStore.update(target.getId(), target.getState(), target.getErrorType(),
+                  target.getErrorMessage());
+            signalStateChange();
+        } catch (BulkStorageException e) {
+            LOGGER.error("{}, updateJobState: {}", ruid, e.toString());
+        }
+    }
 }
diff --git a/modules/dcache-bulk/src/main/java/org/dcache/services/bulk/job/RequestContainerJob.java b/modules/dcache-bulk/src/main/java/org/dcache/services/bulk/job/RequestContainerJob.java
@@ -256,6 +256,11 @@ private ListenableFuture perform(Long id, PID pid, FsPath path, FileAttributes a
             future = Futures.immediateFailedFuture(e);
             register(id, pid, path, future, attributes, e);
             return future;
+        } catch (Throwable e) {
+            future = Futures.immediateFailedFuture(e);
+            register(id, pid, path, future, attributes, e);
+            uncaughtException(Thread.currentThread(), e);
+            return future;
         }
 
         register(id, pid, path, future, attributes, null);