add log for unexpected pipeline error (#5795)

Signed-off-by: Karim TAAM <karim.t2am@gmail.com>

services/pipeline/src/main/java/org/hyperledger/besu/services/pipeline/AsyncOperationProcessor.java

@@ -16,6 +16,8 @@ package org.hyperledger.besu.services.pipeline;
 
 import static java.util.concurrent.CompletableFuture.completedFuture;
 
+import org.hyperledger.besu.services.pipeline.exception.AsyncOperationException;
+
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
@@ -83,7 +85,7 @@ class AsyncOperationProcessor<I, O> implements Processor<I, O> {
     } catch (final InterruptedException e) {
       LOG.trace("Interrupted while waiting for processing to complete", e);
     } catch (final ExecutionException e) {
-      throw new RuntimeException("Async operation failed. " + e.getMessage(), e);
+      throw new AsyncOperationException("Async operation failed. " + e.getMessage(), e);
     } catch (final TimeoutException e) {
       // Ignore and go back around the loop.
     }

services/pipeline/src/main/java/org/hyperledger/besu/services/pipeline/Pipeline.java

@@ -16,12 +16,14 @@ package org.hyperledger.besu.services.pipeline;
 
 import static java.util.stream.Collectors.toList;
 
+import org.hyperledger.besu.services.pipeline.exception.AsyncOperationException;
 import org.hyperledger.besu.util.ExceptionUtils;
 
 import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -169,7 +171,13 @@ public class Pipeline<I> {
             if (tracingEnabled) {
               taskSpan.setStatus(StatusCode.ERROR);
             }
-            LOG.debug("Unhandled exception in pipeline. Aborting.", t);
+            if (t instanceof CompletionException
+                || t instanceof CancellationException
+                || t instanceof AsyncOperationException) {
+              LOG.debug("Unhandled exception in pipeline. Aborting.", t);
+            } else {
+              LOG.info("Unexpected exception in pipeline. Aborting.", t);
+            }
             try {
               abort(t);
             } catch (final Throwable t2) {

services/pipeline/src/main/java/org/hyperledger/besu/services/pipeline/exception/AsyncOperationException.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright Hyperledger Besu Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.services.pipeline.exception;
+
+/** This class allows throwing an exception in case of failure of an async task in the pipeline */
+public class AsyncOperationException extends RuntimeException {
+
+  /**
+   * Constructor of the exception that takes the message and the cause of it.
+   *
+   * @param message of the exception
+   * @param cause of the exception
+   */
+  public AsyncOperationException(final String message, final Throwable cause) {
+    super(message, cause);
+  }
+}