feature: #242 Explicit terminal and retry exceptions for cleaner logging and poison pills

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/RetriableException.java

@@ -0,0 +1,29 @@
+package io.confluent.parallelconsumer;
+
+/*-
+ * Copyright (C) 2020-2022 Confluent, Inc.
+ */
+
+/**
+ * A user's processing function can throw this exception, which signals to PC that processing of the message has failed,
+ * and that it should be retired at a later time.
+ * <p>
+ * The advantage of throwing this exception explicitly, is that PC will not log an ERROR. If any other type of exception
+ * is thrown by the user's function, that will be logged as an error (but will still be retried later).
+ * <p>
+ * So in short, if this exception is thrown, nothing will be logged (except at DEBUG level), any other exception will be
+ * logged as an error.
+ */
+public class RetriableException extends RuntimeException {
+    public RetriableException(String message) {
+        super(message);
+    }
+
+    public RetriableException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public RetriableException(Throwable cause) {
+        super(cause);
+    }
+}

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/AbstractParallelEoSStreamProcessor.java

@@ -8,6 +8,7 @@
 import io.confluent.parallelconsumer.ParallelConsumer;
 import io.confluent.parallelconsumer.ParallelConsumerOptions;
 import io.confluent.parallelconsumer.PollContextInternal;
+import io.confluent.parallelconsumer.RetriableException;
 import io.confluent.parallelconsumer.state.WorkContainer;
 import io.confluent.parallelconsumer.state.WorkManager;
 import lombok.*;
@@ -1121,7 +1122,12 @@ protected <R> List<ParallelConsumer.Tuple<ConsumerRecord<K, V>, R>> runUserFunct
             return intermediateResults;
         } catch (Exception e) {
             // handle fail
-            log.error("Exception caught in user function running stage, registering WC as failed, returning to mailbox", e);
+            String msg = "Exception caught in user function running stage, registering WC as failed, returning to mailbox";
+            if (e instanceof RetriableException) {
+                log.debug("Explicit " + RetriableException.class.getSimpleName() + " caught, logging at DEBUG only. " + msg, e);
+            } else {
+                log.error(msg, e);
+            }
             for (var wc : workContainerBatch) {
                 wc.onUserFunctionFailure(e);
                 addToMailbox(wc); // always add on error

parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/FakeRuntimeError.java

@@ -7,7 +7,7 @@
 /**
  * Used for testing error handling - easier to identify than a plan exception.
  */
-public class FakeRuntimeError extends RuntimeException {
+public class FakeRuntimeError extends RetriableException {
     public FakeRuntimeError(String msg) {
         super(msg);
     }