[FLINK-39540][Connectors/Kinesis][5.1] Addressed bugs for EFO subscriptions when they are completed#245
Open
pelaezryan wants to merge 2 commits into
Open
[FLINK-39540][Connectors/Kinesis][5.1] Addressed bugs for EFO subscriptions when they are completed#245pelaezryan wants to merge 2 commits into
pelaezryan wants to merge 2 commits into
Conversation
davidradl
reviewed
May 5, 2026
| consumerArn); | ||
| if (subscriptionActive.get()) { | ||
| if (shardSubscriber != null | ||
| && shardSubscriber.getSubscriptionState() == SubscriptionState.SUBSCRIBED) { |
There was a problem hiding this comment.
nit: should we change the message. It refers to the subscription being active , which seems to be tied to the subscriptionActive variable we have removed.
davidradl
reviewed
May 5, 2026
davidradl
reviewed
May 5, 2026
|
|
||
| public void cancel() { | ||
| if (!subscriptionActive.get()) { | ||
| if (this.subscriptionState.get() == SubscriptionState.COMPLETED) { |
There was a problem hiding this comment.
change the message to be more accurate?
Contributor
Author
There was a problem hiding this comment.
The log statement is technically correct in that its canceling a subscription that is no longer active. But ill update the wording to more closely align with the new states
davidradl
reviewed
May 5, 2026
| private enum SubscriptionState { | ||
| NOT_STARTED, | ||
| SUBSCRIBED, | ||
| COMPLETED |
Contributor
Author
There was a problem hiding this comment.
Not that I am aware of. When any exceptions occur we terminate the subscription (if it's not done so already):
kinesis.subscribeToShard(consumerArn, shardId, startingPosition, responseHandler)
.exceptionally(
throwable -> {
// If consumer exists and is still activating, we want to countdown.
if (ExceptionUtils.findThrowable(
throwable, ResourceInUseException.class)
.isPresent()) {
waitForSubscriptionLatch.countDown();
return null;
}
LOG.error(
"Error subscribing to shard {} with starting position {} for consumer {}.",
shardId,
startingPosition,
consumerArn,
throwable);
terminateSubscription(throwable);
return null;
});
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose of the change
Address a bug in the Kinesis Flink Connector that would cause Enhanced Fan Out (EFO) customers to continously experience
IllegalArgumentExceptionwhen their stream is resharded. This occurs due to the subscription being restarted when it should be completed and cleaned up.Example of error:
Verifying this change
After making the change, i 1. re-ran my flink application, 2. resharded my stream. Afterwards verified that subscriptions were being shut down and restarted without issues:
reshard
Significant changes
N/A - Not a significant change