From 9ffffacb3e6ac5166665f16ff6a999c7a76ad452 Mon Sep 17 00:00:00 2001
From: Philipp Schirmer <philipp.schirmer@bakdata.com>
Date: Thu, 21 Mar 2024 13:55:31 +0100
Subject: [PATCH] Forward streams errors if cause is recoverable

---
 .../src/main/java/com/bakdata/kafka/ErrorUtil.java   | 12 +++++++++++-
 .../test/java/com/bakdata/kafka/ErrorUtilTest.java   |  5 ++++-
 2 files changed, 15 insertions(+), 2 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
index 070743d..1d778b5 100644
--- a/error-handling-core/src/main/java/com/bakdata/kafka/ErrorUtil.java
+++ b/error-handling-core/src/main/java/com/bakdata/kafka/ErrorUtil.java
@@ -51,6 +51,7 @@
 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";
 
     /**
      * Check if an exception is recoverable and thus should be thrown so that the process is restarted by the execution
@@ -70,6 +71,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.
+     * 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}
@@ -79,9 +82,16 @@ 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())) {
+        final String packageName = e.getClass().getPackageName();
+        if (ORG_APACHE_KAFKA_COMMON_ERRORS.equals(packageName)) {
             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
index d7d7aec..152af2a 100644
--- 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,9 @@ 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(new SerializationException()), true),
+                Arguments.of(new StreamsException("message"), false)
         );
     }