This is an automated email from the ASF dual-hosted git repository. dannycranmer pushed a commit to branch release-1.15 in repository https://gitbox.apache.org/repos/asf/flink.git
commit 9715b5e641e76a33c0b869409ce5094f5727105a Author: Alexander Fedulov <[email protected]> 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 11da7aedc45..72d34ad8271 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 @@ -49,6 +49,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; @@ -93,6 +94,18 @@ public class KinesisProxyTest { assertTrue(kinesisProxy.isRecoverableSdkClientException(ex)); } + @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 =
