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.*;
@@ -45,6 +46,8 @@
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static lombok.AccessLevel.PRIVATE;
 import static lombok.AccessLevel.PROTECTED;
+import static org.slf4j.event.Level.DEBUG;
+import static org.slf4j.event.Level.WARN;
 
 /**
  * @see ParallelConsumer
@@ -1120,8 +1123,10 @@ 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);
+            var level = e instanceof RetriableException ? DEBUG : WARN;
+            var prefix = e instanceof RetriableException ? "Explicit " + RetriableException.class.getSimpleName() + " caught: " : "";
+            log.atLevel(level)
+                    .log(prefix + "Exception caught in user function running stage, registering WC as failed, returning to queue", 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);
     }

pom.xml

@@ -99,7 +99,7 @@
         <surefire.version>3.0.0-M6</surefire.version>
 
         <!-- core -->
-        <slf4j.version>1.7.36</slf4j.version>
+        <slf4j.version>2.0.0-alpha7</slf4j.version>
         <kafka.version>3.1.0</kafka.version>
         <version.unij>0.1.3</version.unij>