Forward streams errors if cause is recoverable

bakdata · Mar 21, 2024 · 72ccda6 · 72ccda6
1 parent 9ffffac
commit 72ccda6
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 13 deletions.
diff --git a/error-handling-core/src/main/java/com/bakdata/kafka/ErrorUtil.java b/error-handling-core/src/main/java/com/bakdata/kafka/ErrorUtil.java
@@ -29,6 +29,7 @@
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
 import java.util.Objects;
+import java.util.Set;
 import lombok.experimental.UtilityClass;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.avro.Schema;
@@ -52,6 +53,10 @@ public class ErrorUtil {
 
     private static final String ORG_APACHE_KAFKA_COMMON_ERRORS = "org.apache.kafka.common.errors";
     private static final String ORG_APACHE_KAFKA_STREAMS_ERRORS = "org.apache.kafka.streams.errors";
+    private static final Set<String> RECOVERABLE_ERROR_PACKAGES = Set.of(
+            ORG_APACHE_KAFKA_COMMON_ERRORS,
+            ORG_APACHE_KAFKA_STREAMS_ERRORS
+    );
 
     /**
      * Check if an exception is recoverable and thus should be thrown so that the process is restarted by the execution
@@ -69,10 +74,9 @@ public static boolean isRecoverable(final Exception e) {
     }
 
     /**
-     * Check if an exception is thrown by Kafka, i.e., located in package {@code org.apache.kafka.common.errors}, and is
+     * Check if an exception is thrown by Kafka, i.e., located in package {@code org.apache.kafka.common.errors} or
+     * {@code org.apache.kafka.streams.errors}, and is
      * recoverable.
-     * If an exception is thrown by Kafka Streams, i.e., located in package {@code org.apache.kafka.streams.errors},
-     * the cause is checked for recoverability.
      * <p>Non-recoverable Kafka errors are:
      * <ul>
      *     <li>{@link RecordTooLargeException}
@@ -82,16 +86,9 @@ public static boolean isRecoverable(final Exception e) {
      * @return whether exception is thrown by Kafka and recoverable or not
      */
     public static boolean isRecoverableKafkaError(final Exception e) {
-        final String packageName = e.getClass().getPackageName();
-        if (ORG_APACHE_KAFKA_COMMON_ERRORS.equals(packageName)) {
+        if (RECOVERABLE_ERROR_PACKAGES.contains(e.getClass().getPackageName())) {
             return !(e instanceof RecordTooLargeException);
         }
-        if (ORG_APACHE_KAFKA_STREAMS_ERRORS.equals(packageName)) {
-            final Throwable cause = e.getCause();
-            if (cause instanceof Exception) {
-                return isRecoverable((Exception) cause);
-            }
-        }
         return false;
     }
 

diff --git a/error-handling-core/src/test/java/com/bakdata/kafka/ErrorUtilTest.java b/error-handling-core/src/test/java/com/bakdata/kafka/ErrorUtilTest.java
@@ -49,8 +49,7 @@ static Stream<Arguments> generateIsRecoverableExceptionParameters() {
         return Stream.of(
                 Arguments.of(mock(Exception.class), false),
                 Arguments.of(new SerializationException(), true),
-                Arguments.of(new StreamsException(new SerializationException()), true),
-                Arguments.of(new StreamsException("message"), false)
+                Arguments.of(new StreamsException("message"), true)
         );
     }