[FLINK-35815][Connector/Kinesis] Fix detection of recoverable exceptions for EFO operations

Author: z3d1k

flink-connector-aws/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/AwsV2Util.java

@@ -18,6 +18,7 @@
 package org.apache.flink.streaming.connectors.kinesis.util;
 
 import org.apache.flink.annotation.Internal;
+import org.apache.flink.util.ExceptionUtils;
 
 import software.amazon.awssdk.http.SdkHttpConfigurationOption;
 import software.amazon.awssdk.services.kinesis.model.LimitExceededException;
@@ -74,8 +75,12 @@ public static boolean isNoneEfoRegistrationType(final Properties properties) {
     }
 
     public static boolean isRecoverableException(Exception e) {
-        Throwable cause = e.getCause();
-        return cause instanceof LimitExceededException
-                || cause instanceof ProvisionedThroughputExceededException;
+        return ExceptionUtils.findThrowable(
+                        e,
+                        throwable ->
+                                throwable instanceof LimitExceededException
+                                        || throwable
+                                                instanceof ProvisionedThroughputExceededException)
+                .isPresent();
     }
 }

flink-connector-aws/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/util/AwsV2UtilTest.java

@@ -20,6 +20,7 @@
 import org.junit.Test;
 import software.amazon.awssdk.http.SdkHttpConfigurationOption;
 import software.amazon.awssdk.services.kinesis.model.LimitExceededException;
+import software.amazon.awssdk.services.kinesis.model.ProvisionedThroughputExceededException;
 import software.amazon.awssdk.utils.AttributeMap;
 
 import java.time.Duration;
@@ -133,7 +134,19 @@ public void testIsNoneEfoRegistrationType() {
     }
 
     @Test
-    public void testIsRecoverableExceptionForRecoverable() {
+    public void testIsRecoverableExceptionForRecoverableLimitExceeded() {
+        Exception recoverable = LimitExceededException.builder().build();
+        assertThat(AwsV2Util.isRecoverableException(recoverable)).isTrue();
+    }
+
+    @Test
+    public void testIsRecoverableExceptionForRecoverableProvisionedThroughputExceeded() {
+        Exception recoverable = ProvisionedThroughputExceededException.builder().build();
+        assertThat(AwsV2Util.isRecoverableException(recoverable)).isTrue();
+    }
+
+    @Test
+    public void testIsRecoverableExceptionForRecoverableWrapped() {
         Exception recoverable = LimitExceededException.builder().build();
         assertThat(AwsV2Util.isRecoverableException(new ExecutionException(recoverable))).isTrue();
     }