Forward streams errors (#27)

bakdata · Mar 21, 2024 · 80338b7 · 80338b7
1 parent 093347a
commit 80338b7
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 4 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;
@@ -51,6 +52,11 @@
 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
@@ -68,8 +74,8 @@ 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
-     * recoverable.
+     * 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.
      * <p>Non-recoverable Kafka errors are:
      * <ul>
      *     <li>{@link RecordTooLargeException}
@@ -79,7 +85,7 @@ 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) {
-        if (ORG_APACHE_KAFKA_COMMON_ERRORS.equals(e.getClass().getPackageName())) {
+        if (RECOVERABLE_ERROR_PACKAGES.contains(e.getClass().getPackageName())) {
             return !(e instanceof RecordTooLargeException);
         }
         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
@@ -29,6 +29,7 @@
 
 import java.util.stream.Stream;
 import org.apache.kafka.common.errors.SerializationException;
+import org.apache.kafka.streams.errors.StreamsException;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
@@ -47,7 +48,8 @@ static Stream<Arguments> generateConvertToStringParameters() {
     static Stream<Arguments> generateIsRecoverableExceptionParameters() {
         return Stream.of(
                 Arguments.of(mock(Exception.class), false),
-                Arguments.of(new SerializationException(), true)
+                Arguments.of(new SerializationException(), true),
+                Arguments.of(new StreamsException("message"), true)
         );
     }