[flink] branch master updated: [FLINK-29476][connector][kinesis] Retry connection termination with EOFException

[dannycranmer]

This is an automated email from the ASF dual-hosted git repository.

dannycranmer pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/master by this push:
     new 49e0041042e [FLINK-29476][connector][kinesis] Retry connection 
termination with EOFException
49e0041042e is described below

commit 49e0041042e65e0045701cc5514ce55d74796ba2
Author: Alexander Fedulov <1492164+afedu...@users.noreply.github.com>
AuthorDate: Thu Sep 29 15:19:23 2022 +0000

    [FLINK-29476][connector][kinesis] Retry connection termination with 
EOFException
---
 .../streaming/connectors/kinesis/proxy/KinesisProxy.java    |  8 +++++++-
 .../connectors/kinesis/proxy/KinesisProxyTest.java          | 13 +++++++++++++
 2 files changed, 20 insertions(+), 1 deletion(-)

diff --git 
a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxy.java
 
b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxy.java
index 641027c98ef..458c02925b5 100644
--- 
a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxy.java
+++ 
b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxy.java
@@ -51,6 +51,7 @@ import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nullable;
 
+import java.io.EOFException;
 import java.net.SocketTimeoutException;
 import java.util.ArrayList;
 import java.util.Date;
@@ -422,13 +423,18 @@ public class KinesisProxy implements 
KinesisProxyInterface {
     protected boolean isRecoverableSdkClientException(SdkClientException ex) {
         if (ex instanceof AmazonServiceException) {
             return 
KinesisProxy.isRecoverableException((AmazonServiceException) ex);
-        } else if (ExceptionUtils.findThrowable(ex, 
SocketTimeoutException.class).isPresent()) {
+        } else if (isRecoverableConnectionException(ex)) {
             return true;
         }
         // customizations may decide to retry other errors, such as read 
timeouts
         return false;
     }
 
+    private boolean isRecoverableConnectionException(SdkClientException ex) {
+        return ExceptionUtils.findThrowable(ex, 
SocketTimeoutException.class).isPresent()
+                || ExceptionUtils.findThrowable(ex, 
EOFException.class).isPresent();
+    }
+
     /**
      * Determines whether the exception is recoverable using 
exponential-backoff.
      *
diff --git 
a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxyTest.java
 
b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxyTest.java
index 01654a4bda9..099742ca56e 100644
--- 
a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxyTest.java
+++ 
b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxyTest.java
@@ -48,6 +48,7 @@ import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 import org.powermock.reflect.Whitebox;
 
+import java.io.EOFException;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.util.ArrayList;
@@ -88,6 +89,18 @@ public class KinesisProxyTest {
         assertThat(kinesisProxy.isRecoverableSdkClientException(ex)).isTrue();
     }
 
+    @Test
+    public void testIsRecoverableExceptionWithEOFConnectError() throws 
UnknownHostException {
+        Properties kinesisConsumerConfig = new Properties();
+        kinesisConsumerConfig.setProperty(ConsumerConfigConstants.AWS_REGION, 
"us-east-1");
+        KinesisProxy kinesisProxy = new KinesisProxy(kinesisConsumerConfig);
+        final SdkClientException ex =
+                new SdkClientException(
+                        "Unable to execute HTTP request",
+                        new EOFException("SSL peer shut down incorrectly"));
+        assertThat(kinesisProxy.isRecoverableSdkClientException(ex)).isTrue();
+    }
+
     @Test
     public void testIsRecoverableExceptionWithProvisionedThroughputExceeded() {
         final ProvisionedThroughputExceededException ex =