Skip to content

Flaky integ test where the SubscribeToShardIntegrationTest #5949

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Mar 12, 2025
Merged

Flaky integ test where the SubscribeToShardIntegrationTest #5949

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d65d8c3
Flaky integ test where the SubscribeToShardIntegrationTest
joviegas Mar 11, 2025
44e2d0e
Merge branch 'master' into joviegas/SubscribeToShardIntegrationTest_I…
joviegas Mar 11, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 16 additions & 1 deletion .../src/it/java/software/amazon/awssdk/services/kinesis/SubscribeToShardIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@
import software.amazon.awssdk.services.kinesis.model.ConsumerStatus;
import software.amazon.awssdk.services.kinesis.model.PutRecordRequest;
import software.amazon.awssdk.services.kinesis.model.Record;
import software.amazon.awssdk.services.kinesis.model.ResourceInUseException;
import software.amazon.awssdk.services.kinesis.model.ShardIteratorType;
import software.amazon.awssdk.services.kinesis.model.StreamStatus;
import software.amazon.awssdk.services.kinesis.model.SubscribeToShardEvent;
Expand Down Expand Up @@ -290,8 +291,22 @@ private static void waitForStreamToBeActive() {
Waiter.run(() -> asyncClient.describeStream(r -> r.streamName(streamName)).join())
.until(b -> b.streamDescription().streamStatus().equals(StreamStatus.ACTIVE))
.orFailAfter(Duration.ofMinutes(5));
}

// Additional verification to ensure stream is fully operational
Waiter.run(() -> {
try {
asyncClient.listShards(r -> r.streamName(streamName)).join();
return true;
} catch (Exception e) {
if (e.getCause() instanceof ResourceInUseException) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We already have a Waiter waiting for the ACTIVE state, but we are failing because the stream is in the CREATING stage? That seems weird, are we sure this is helping?

Copy link
Contributor Author

@joviegas joviegas Mar 11, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeap , I have attached the complete log in the internal issue.
The above log is taken from the codebuild logs.
I have started Integ test too as part of the codebuild

Copy link
Contributor Author

@joviegas joviegas Mar 11, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The additional check will ensure that the actual state is properly reflected in the system and will provide additional time with retry logic rather than immediately failing(same as we retrying the test case when it fails)
I recommend proceeding with this safe approach since the current failure of this test case is unnecessarily blocking the release and creating significant churn.

return false;
}
throw e;
}
})
.until(Boolean::booleanValue)
.orFailAfter(Duration.ofMinutes(1));
}

/**
* Puts a random record to the stream.
Expand Down
Loading