Kds stream api changes #5239
Kds stream api changes #5239
Conversation
Signed-off-by: Souvik Bose <souvbose@amazon.com>
Signed-off-by: Souvik Bose <souvbose@amazon.com>
Signed-off-by: Souvik Bose <souvbose@amazon.com>
sb2k16
requested review from
chenqi0805,
engechas,
graytaylor0,
dinujoh,
kkondaka,
KarstenSchnitter,
dlvenable and
oeyh
as code owners
| import java.time.Duration; | ||
|
|
||
| @Slf4j | ||
| public class KinesisClientAPIHandler { |
There was a problem hiding this comment.
Rename to KinesisClientApiHandler to promote consistency.
| failedAttemptCount = 0; | ||
| while (failedAttemptCount < maxRetryCount) { | ||
| try { | ||
| DescribeStreamSummaryRequest describeStreamSummaryRequest = DescribeStreamSummaryRequest.builder() |
There was a problem hiding this comment.
You can move this outside of the while loop so that you don't have to recreate it.
| try { | ||
| DescribeStreamSummaryRequest describeStreamSummaryRequest = DescribeStreamSummaryRequest.builder() | ||
| .streamName(streamName).build(); | ||
| DescribeStreamSummaryResponse response = kinesisClient.describeStreamSummary(describeStreamSummaryRequest).join(); |
There was a problem hiding this comment.
Does this require different kinesis: permissions?
There was a problem hiding this comment.
Thanks @dlvenable . Yes, we will need to different kinesis:DescribeStreamSummary permissions. This is already documented as KCL requires it.
| DescribeStreamSummaryResponse response = kinesisClient.describeStreamSummary(describeStreamSummaryRequest).join(); | ||
| String streamIdentifierString = getStreamIdentifierString(response.streamDescriptionSummary()); | ||
| return StreamIdentifier.multiStreamInstance(streamIdentifierString); | ||
| } catch (Exception ex) { |
There was a problem hiding this comment.
Let's catch some known exceptions that the Kinesis client throws for server or client errors and log those without the stack trace.
e.g.
catch (GenericKinesisException ex) {
log.error("Failed to describe stream summary for stream {} with error {}. The kinesis source will retry.",
streamName, ex.getMessage());
} catch (GenericKinesisException ex) {
log.error("Failed to describe stream summary for stream {}. The kinesis source will retry.",
streamName, ex);
}
| private void applyBackoff() { | ||
| final long delayMillis = backoff.nextDelayMillis(failedAttemptCount); | ||
| if (delayMillis < 0) { | ||
| Thread.currentThread().interrupt(); |
There was a problem hiding this comment.
Why are you interrupting the current thread?
| try { | ||
| Thread.sleep(delayMillis); | ||
| } catch (final InterruptedException e){ | ||
| log.error("Thread is interrupted while polling SQS with retry.", e); |
There was a problem hiding this comment.
| log.error("Thread is interrupted while polling SQS with retry.", e); | |
| log.error("Thread is interrupted while polling Kinesis with retry.", e); |
| import static org.mockito.Mockito.when; | ||
|
|
||
| public class KinesisClientAPIHandlerTest { | ||
| private static final List<String> STREAMS_LIST = ImmutableList.of("stream-1", "stream-2", "stream-3"); |
There was a problem hiding this comment.
I'd recommend making this a variable you initialize in @BeforeEach. Also, you could even have @Nested classes to test the scenario of 0 and 1 streams as well.
Also, you can use Java's List.of() which is also immutable.
Signed-off-by: Souvik Bose <souvbose@amazon.com>
* Change API to use DescribeStreamSummary Signed-off-by: Souvik Bose <souvbose@amazon.com> * Code changes Signed-off-by: Souvik Bose <souvbose@amazon.com> * Rename the class. Signed-off-by: Souvik Bose <souvbose@amazon.com> * Address review comments Signed-off-by: Souvik Bose <souvbose@amazon.com> --------- Signed-off-by: Souvik Bose <souvbose@amazon.com> Co-authored-by: Souvik Bose <souvbose@amazon.com>
Description
This PR is to use the
DescribeStreamSummaryAPI to get the stream identifier instead of theDescribeStream. It also includes handling retries using an exponential backoff strategy.Issues Resolved
Resolves #1082
Check List
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.