You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ja...@apache.org on 2026/07/09 06:04:20 UTC
(camel-quarkus) branch main updated: Fixes #7733. Fix Aws2KinesisTest.failingDefaultCredentialsProviderTest flakiness
This is an automated email from the ASF dual-hosted git repository.
jamesnetherton pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-quarkus.git
The following commit(s) were added to refs/heads/main by this push:
new 9947fdb84f Fixes #7733. Fix Aws2KinesisTest.failingDefaultCredentialsProviderTest flakiness
9947fdb84f is described below
commit 9947fdb84f123a2c636b2eab8c008487dc4c3631
Author: James Netherton <ja...@users.noreply.github.com>
AuthorDate: Thu Jul 9 07:04:14 2026 +0100
Fixes #7733. Fix Aws2KinesisTest.failingDefaultCredentialsProviderTest flakiness
Suspend Kinesis consumer routes before resetting the shared
KinesisConnection to prevent a race where a consumer poll thread
recreates the client with static credentials before the producer
can recreate it with DefaultCredentialsProvider.
Co-authored-by: Claude Opus 4.6 <no...@anthropic.com>
---
.../component/aws2/kinesis/it/Aws2KinesisResource.java | 18 +++++++++++++++++-
1 file changed, 17 insertions(+), 1 deletion(-)
diff --git a/integration-test-groups/aws2/aws2-kinesis/src/main/java/org/apache/camel/quarkus/component/aws2/kinesis/it/Aws2KinesisResource.java b/integration-test-groups/aws2/aws2-kinesis/src/main/java/org/apache/camel/quarkus/component/aws2/kinesis/it/Aws2KinesisResource.java
index e29fedc396..fa5b430eba 100644
--- a/integration-test-groups/aws2/aws2-kinesis/src/main/java/org/apache/camel/quarkus/component/aws2/kinesis/it/Aws2KinesisResource.java
+++ b/integration-test-groups/aws2/aws2-kinesis/src/main/java/org/apache/camel/quarkus/component/aws2/kinesis/it/Aws2KinesisResource.java
@@ -104,11 +104,27 @@ public class Aws2KinesisResource extends BaseAws2Resource {
@Override
protected void onDefaultCredentialsProviderChange() throws Exception {
log.info("onDefaultCredentialsProviderChange start");
- //reset connection, because irt is cached since https://github.com/apache/camel/pull/10919
+
+ // Suspend consumers so they don't race to recreate the shared KinesisClient
+ // with static credentials before the producer can recreate it with DefaultCredentialsProvider
+ for (org.apache.camel.Route route : camelContext.getRoutes()) {
+ if (route.getEndpoint().getEndpointUri().startsWith("aws2-kinesis://")) {
+ camelContext.getRouteController().suspendRoute(route.getRouteId());
+ }
+ }
+
+ //reset connection, because it is cached since https://github.com/apache/camel/pull/10919
KinesisConnection kc = camelContext.getRegistry().findSingleByType(Kinesis2Component.class).getConnection();
kc.close();
kc.setKinesisAsyncClient(null);
kc.setKinesisClient(null);
super.onDefaultCredentialsProviderChange();
+
+ // Resume consumers
+ for (org.apache.camel.Route route : camelContext.getRoutes()) {
+ if (route.getEndpoint().getEndpointUri().startsWith("aws2-kinesis://")) {
+ camelContext.getRouteController().resumeRoute(route.getRouteId());
+ }
+ }
}
}