[FLINK-5355] Handle AmazonKinesisException gracefully in Kinesis Streaming Connector #3078

skidder · 2017-01-06T18:21:05Z

My Flink job that consumes from a Kinesis stream must be restarted at least once daily due to an uncaught AmazonKinesisException when reading from Kinesis. The complete stacktrace looks like:

com.amazonaws.services.kinesis.model.AmazonKinesisException: null (Service: AmazonKinesis; Status Code: 500; Error Code: InternalFailure; Request ID: dc1b7a1a-1b97-1a32-8cd5-79a896a55223)
	at com.amazonaws.http.AmazonHttpClient$RequestExecutor.handleErrorResponse(AmazonHttpClient.java:1545)
	at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeOneRequest(AmazonHttpClient.java:1183)
	at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeHelper(AmazonHttpClient.java:964)
	at com.amazonaws.http.AmazonHttpClient$RequestExecutor.doExecute(AmazonHttpClient.java:676)
	at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeWithTimer(AmazonHttpClient.java:650)
	at com.amazonaws.http.AmazonHttpClient$RequestExecutor.execute(AmazonHttpClient.java:633)
	at com.amazonaws.http.AmazonHttpClient$RequestExecutor.access$300(AmazonHttpClient.java:601)
	at com.amazonaws.http.AmazonHttpClient$RequestExecutionBuilderImpl.execute(AmazonHttpClient.java:583)
	at com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:447)
	at com.amazonaws.services.kinesis.AmazonKinesisClient.doInvoke(AmazonKinesisClient.java:1747)
	at com.amazonaws.services.kinesis.AmazonKinesisClient.invoke(AmazonKinesisClient.java:1723)
	at com.amazonaws.services.kinesis.AmazonKinesisClient.getRecords(AmazonKinesisClient.java:858)
	at org.apache.flink.streaming.connectors.kinesis.proxy.KinesisProxy.getRecords(KinesisProxy.java:193)
	at org.apache.flink.streaming.connectors.kinesis.internals.ShardConsumer.getRecords(ShardConsumer.java:268)
	at org.apache.flink.streaming.connectors.kinesis.internals.ShardConsumer.run(ShardConsumer.java:176)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	at java.lang.Thread.run(Thread.java:745)

It's interesting that the Kinesis endpoint returned a 500 status code, but that's outside the scope of this issue.

I think we can handle this exception in the same manner as a ProvisionedThroughputException: performing an exponential backoff and retrying a finite number of times before throwing an exception.

…aming Connector

tzulitai · 2017-01-08T11:28:50Z

Thank you for opening a pull request for this @skidder !

I think AmazonServiceException will also be thrown for client errors (ex., wrong user provided parameters for API calls, such as invalid access key). Whether an exception is caused by client or server is indicated by AmazonServiceException#getErrorType() which returns either ErrorType.Client, ErrorType.Server, or ErrorType.Unknown

So, instead of simply catching a AmazonServiceException, I wonder if we should only perform exponential backoff for service exceptions with Server and Unknown error types. We need to also see what type ProvisionedThroughputExceededException is, and see if it is included as a Server or Unknown; if not, handle that also.

What do you think?

skidder · 2017-01-09T21:21:01Z

Thanks @tzulitai , great feedback!

The ProvisionedThroughputExceededException exception will be reported with an HTTP 400 response status code:
http:https://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetRecords.html#API_GetRecords_Errors

The AWS SDK will assign the ErrorType.Client type to all exceptions with an HTTP status code less than 500 (AWS SDK source).

Perhaps we can perform exponential-backoff for exceptions where:

Client error of type ProvisionedThroughputExceededException
All Service errors (e.g. HTTP 500, 503)
All Unknown errors (appear to be limited to errors unmarshalling the Kinesis service response)

All other exceptions can be thrown up.

What are your thoughts?

tzulitai

Hi @skidder,
Really sorry for the late response, this PR somehow slipped out of my attention (please feel free to ping me if I'm apparently not responding in the future :))

Thank you for the detailed explanation! The proposal seems very reasonable to me.

I have some last minor comments on the code style consistency, once those are addressed the code is ready to be merged :)

tzulitai · 2017-01-15T19:36:56Z

...-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxy.java

+ * exponential-backoff
+ * 
+ * @param ex
+ * Exception to inspect


nit: I think the Javadoc formatting here is inconsistent with the other methods (line change).

tzulitai · 2017-01-15T19:37:38Z

...-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxy.java

+ return false;
+ }
+
+ switch (ex.getErrorType()) {


The indentation for the cases here seem to be missing.

tzulitai · 2017-01-15T19:38:19Z

...-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxy.java

+
+ switch (ex.getErrorType()) {
+ case Client:
+ if (ex instanceof ProvisionedThroughputExceededException) {


It'll probably be cleaner to just do ex instanceof ProvisionedThroughputExceededException

tzulitai · 2017-01-15T19:38:55Z

...esis/src/test/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxyTest.java

+
+/**
+ * Test for methods in the KinesisProxy class.
+ * 


Extra empty line

tzulitai · 2017-01-15T19:39:21Z

...esis/src/test/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxyTest.java

+import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException;
+
+/**
+ * Test for methods in the KinesisProxy class.


Should link the KinesisProxy referencing, like other Javadocs.

tzulitai · 2017-01-15T19:39:57Z

...esis/src/test/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxyTest.java

+
+package org.apache.flink.streaming.connectors.kinesis.proxy;
+
+import static org.junit.Assert.*;


In Flink we generally try to avoid asterisk imports.
The style check doesn't actually check the test scope, but it'll be good to try to be consistent with the style rules in tests also.

skidder · 2017-01-19T17:48:43Z

Thank you @tzulitai for the feedback on the styling! I've pushed a commit that addresses your comments.

tzulitai · 2017-01-20T15:24:48Z

Thank you @skidder for addressing the comments and the contribution! Merging ..

…nesis Streaming Connector This closes #3078.

…nesis Streaming Connector This closes apache#3078.

[FLINK-5355] Handle AmazonKinesisException gracefully in Kinesis Stre…

85adf13

…aming Connector

skidder added 3 commits January 9, 2017 16:13

[FLINK-5355] Distinguish between AmazonServiceException error-types

19cb47c

[FLINK-5355] Add Apache License header to KinesisProxyTest file

9aa58ae

[FLINK-5355] Add braces to meet checkstyle requirements

3c10301

tzulitai requested changes Jan 15, 2017

View reviewed changes

[FLINK-5355] Fix code style issues

4bd4f9d

asfgit closed this in b380bd3 Jan 20, 2017

asfgit pushed a commit that referenced this pull request Jan 20, 2017

[FLINK-5355] [kinesis] Handle AmazonKinesisException gracefully in Ki…

3b5882a

…nesis Streaming Connector This closes #3078.

alpinegizmo pushed a commit to alpinegizmo/flink that referenced this pull request Jan 20, 2017

[FLINK-5355] [kinesis] Handle AmazonKinesisException gracefully in Ki…

3e9ada7

…nesis Streaming Connector This closes apache#3078.

joseprupi pushed a commit to joseprupi/flink that referenced this pull request Feb 12, 2017

[FLINK-5355] [kinesis] Handle AmazonKinesisException gracefully in Ki…

5946572

…nesis Streaming Connector This closes apache#3078.

rmetzger added the component=Connectors/Kinesis label Mar 14, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-5355] Handle AmazonKinesisException gracefully in Kinesis Streaming Connector #3078

[FLINK-5355] Handle AmazonKinesisException gracefully in Kinesis Streaming Connector #3078

skidder commented Jan 6, 2017

tzulitai commented Jan 8, 2017 •

edited

Loading

skidder commented Jan 9, 2017 •

edited

Loading

tzulitai left a comment

tzulitai Jan 15, 2017

tzulitai Jan 15, 2017

tzulitai Jan 15, 2017

tzulitai Jan 15, 2017

tzulitai Jan 15, 2017

tzulitai Jan 15, 2017

skidder commented Jan 19, 2017

tzulitai commented Jan 20, 2017 •

edited

Loading


		package org.apache.flink.streaming.connectors.kinesis.proxy;

		import static org.junit.Assert.*;

[FLINK-5355] Handle AmazonKinesisException gracefully in Kinesis Streaming Connector #3078

[FLINK-5355] Handle AmazonKinesisException gracefully in Kinesis Streaming Connector #3078

Conversation

skidder commented Jan 6, 2017

tzulitai commented Jan 8, 2017 • edited Loading

skidder commented Jan 9, 2017 • edited Loading

tzulitai left a comment

Choose a reason for hiding this comment

tzulitai Jan 15, 2017

Choose a reason for hiding this comment

tzulitai Jan 15, 2017

Choose a reason for hiding this comment

tzulitai Jan 15, 2017

Choose a reason for hiding this comment

tzulitai Jan 15, 2017

Choose a reason for hiding this comment

tzulitai Jan 15, 2017

Choose a reason for hiding this comment

tzulitai Jan 15, 2017

Choose a reason for hiding this comment

skidder commented Jan 19, 2017

tzulitai commented Jan 20, 2017 • edited Loading

tzulitai commented Jan 8, 2017 •

edited

Loading

skidder commented Jan 9, 2017 •

edited

Loading

tzulitai commented Jan 20, 2017 •

edited

Loading