This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 4ee3a7e1dd08 CAMEL-25909: camel-aws2-kinesis - each endpoint uses and
closes its own client (#28433)
4ee3a7e1dd08 is described below
commit 4ee3a7e1dd08af911dd1dd51ed1f7dec1f75f15c
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Oct 25 13:01:42 2026 +0530
CAMEL-25909: camel-aws2-kinesis - each endpoint uses and closes its own
client (#28433)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../component/aws2/kinesis/Kinesis2Component.java | 4 +
.../component/aws2/kinesis/Kinesis2Endpoint.java | 21 ++-
.../component/aws2/kinesis/KinesisConnection.java | 47 +++++--
.../aws2/kinesis/KinesisEndpointClientTest.java | 151 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 10 ++
5 files changed, 210 insertions(+), 23 deletions(-)
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Component.java
b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Component.java
index aef81b12aba6..765dbbb32749 100644
---
a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Component.java
+++
b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Component.java
@@ -82,6 +82,10 @@ public class Kinesis2Component extends HealthCheckComponent {
IOHelper.close(connection);
}
+ /**
+ * The connection with the client set on this component (if any). The
endpoints do not use it: each endpoint has a
+ * connection of its own, with the client of its own configuration.
+ */
public KinesisConnection getConnection() {
return connection;
}
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Endpoint.java
b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Endpoint.java
index 85746c41217b..f4a299963947 100644
---
a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Endpoint.java
+++
b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Endpoint.java
@@ -62,6 +62,9 @@ public class Kinesis2Endpoint extends ScheduledPollEndpoint
implements EndpointS
@UriParam
private Kinesis2Configuration configuration;
+ // the clients of this endpoint: a client created from the configuration
of another endpoint has another region or
+ // other credentials, and stopping another endpoint must not close the
client of this one
+ private final KinesisConnection kinesisConnection = new
KinesisConnection();
private KinesisClient kinesisClient;
private KinesisAsyncClient kinesisAsyncClient;
@@ -74,8 +77,6 @@ public class Kinesis2Endpoint extends ScheduledPollEndpoint
implements EndpointS
protected void doStart() throws Exception {
super.doStart();
- var kinesisConnection = getComponent().getConnection();
-
if (!configuration.isCborEnabled()) {
System.setProperty(CBOR_ENABLED.property(), "false");
}
@@ -97,13 +98,11 @@ public class Kinesis2Endpoint extends ScheduledPollEndpoint
implements EndpointS
@Override
public void doStop() throws Exception {
- if (ObjectHelper.isEmpty(configuration.getAmazonKinesisClient())) {
- if (ObjectHelper.isNotEmpty(kinesisClient)) {
- kinesisClient.close();
- } else if (ObjectHelper.isNotEmpty(kinesisAsyncClient)) {
- kinesisAsyncClient.close();
- }
- }
+ // closes only the clients this endpoint created, not a client given
with amazonKinesisClient or
+ // amazonKinesisAsyncClient
+ kinesisConnection.close();
+ kinesisClient = null;
+ kinesisAsyncClient = null;
if (!configuration.isCborEnabled()) {
System.clearProperty(CBOR_ENABLED.property());
}
@@ -113,7 +112,7 @@ public class Kinesis2Endpoint extends ScheduledPollEndpoint
implements EndpointS
@Override
public Producer createProducer() throws Exception {
Kinesis2Producer producer = new Kinesis2Producer(this);
- producer.setConnection(getComponent().getConnection());
+ producer.setConnection(kinesisConnection);
return producer;
}
@@ -121,7 +120,7 @@ public class Kinesis2Endpoint extends ScheduledPollEndpoint
implements EndpointS
public Consumer createConsumer(Processor processor) throws Exception {
if (!getConfiguration().isUseKclConsumers()) {
final Kinesis2Consumer consumer = new Kinesis2Consumer(this,
processor);
- consumer.setConnection(getComponent().getConnection());
+ consumer.setConnection(kinesisConnection);
consumer.setSchedulerProperties(getSchedulerProperties());
configureConsumer(consumer);
return consumer;
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/KinesisConnection.java
b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/KinesisConnection.java
index 64d32eaf34a1..7ac848eb1f41 100644
---
a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/KinesisConnection.java
+++
b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/KinesisConnection.java
@@ -33,15 +33,21 @@ public class KinesisConnection implements Closeable {
private KinesisClient kinesisClient;
private KinesisAsyncClient kinesisAsyncClient;
+ // whether the clients were created here (and are closed here), rather
than given by the user
+ private boolean kinesisClientCreated;
+ private boolean kinesisAsyncClientCreated;
private final Lock lock = new ReentrantLock();
public KinesisClient getClient(final Kinesis2Endpoint endpoint) {
lock.lock();
try {
if (ObjectHelper.isEmpty(kinesisClient)) {
- kinesisClient =
ObjectHelper.isNotEmpty(endpoint.getConfiguration().getAmazonKinesisClient())
- ? endpoint.getConfiguration().getAmazonKinesisClient()
- :
KinesisClientFactory.getKinesisClient(endpoint.getConfiguration());
+ if
(ObjectHelper.isNotEmpty(endpoint.getConfiguration().getAmazonKinesisClient()))
{
+ kinesisClient =
endpoint.getConfiguration().getAmazonKinesisClient();
+ } else {
+ kinesisClient =
KinesisClientFactory.getKinesisClient(endpoint.getConfiguration());
+ kinesisClientCreated = true;
+ }
}
return kinesisClient;
} finally {
@@ -53,9 +59,12 @@ public class KinesisConnection implements Closeable {
lock.lock();
try {
if (ObjectHelper.isEmpty(kinesisAsyncClient)) {
- kinesisAsyncClient =
ObjectHelper.isNotEmpty(endpoint.getConfiguration().getAmazonKinesisAsyncClient())
- ?
endpoint.getConfiguration().getAmazonKinesisAsyncClient()
- :
KinesisClientFactory.getKinesisAsyncClient(endpoint.getConfiguration());
+ if
(ObjectHelper.isNotEmpty(endpoint.getConfiguration().getAmazonKinesisAsyncClient()))
{
+ kinesisAsyncClient =
endpoint.getConfiguration().getAmazonKinesisAsyncClient();
+ } else {
+ kinesisAsyncClient =
KinesisClientFactory.getKinesisAsyncClient(endpoint.getConfiguration());
+ kinesisAsyncClientCreated = true;
+ }
}
return kinesisAsyncClient;
} finally {
@@ -64,25 +73,39 @@ public class KinesisConnection implements Closeable {
}
public void setKinesisClient(final KinesisClient kinesisClient) {
- this.kinesisClient = kinesisClient;
+ lock.lock();
+ try {
+ this.kinesisClient = kinesisClient;
+ this.kinesisClientCreated = false;
+ } finally {
+ lock.unlock();
+ }
}
public void setKinesisAsyncClient(final KinesisAsyncClient
kinesisAsyncClient) {
- this.kinesisAsyncClient = kinesisAsyncClient;
+ lock.lock();
+ try {
+ this.kinesisAsyncClient = kinesisAsyncClient;
+ this.kinesisAsyncClientCreated = false;
+ } finally {
+ lock.unlock();
+ }
}
@Override
public void close() throws IOException {
lock.lock();
try {
- if (ObjectHelper.isNotEmpty(kinesisClient)) {
+ if (kinesisClientCreated &&
ObjectHelper.isNotEmpty(kinesisClient)) {
kinesisClient.close();
- kinesisClient = null;
}
- if (ObjectHelper.isNotEmpty(kinesisAsyncClient)) {
+ if (kinesisAsyncClientCreated &&
ObjectHelper.isNotEmpty(kinesisAsyncClient)) {
kinesisAsyncClient.close();
- kinesisAsyncClient = null;
}
+ kinesisClient = null;
+ kinesisAsyncClient = null;
+ kinesisClientCreated = false;
+ kinesisAsyncClientCreated = false;
} finally {
lock.unlock();
}
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisEndpointClientTest.java
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisEndpointClientTest.java
new file mode 100644
index 000000000000..69ce67583c3d
--- /dev/null
+++
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisEndpointClientTest.java
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.services.kinesis.KinesisClient;
+import software.amazon.awssdk.services.kinesis.model.PutRecordRequest;
+import software.amazon.awssdk.services.kinesis.model.PutRecordResponse;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+
+/**
+ * Each aws2-kinesis endpoint uses the client of its own configuration, and
stopping one endpoint does not close the
+ * client of another.
+ */
+public class KinesisEndpointClientTest {
+
+ // nothing listens there: a call with an open client fails to connect, a
call with a closed client fails at once
+ private static final String NO_SERVER =
"&overrideEndpoint=true&uriEndpointOverride=http://localhost:1";;
+ private static final String STREAM_A =
"aws2-kinesis:streamA?accessKey=a&secretKey=a®ion=us-east-1" + NO_SERVER;
+ private static final String STREAM_B =
"aws2-kinesis:streamB?accessKey=b&secretKey=b®ion=eu-west-1" + NO_SERVER;
+
+ private CamelContext context;
+
+ @BeforeEach
+ public void createContext() {
+ context = new DefaultCamelContext();
+ }
+
+ @AfterEach
+ public void stopContext() {
+ context.stop();
+ }
+
+ @Test
+ public void eachEndpointCreatesTheClientOfItsOwnConfiguration() throws
Exception {
+ addRoutes(STREAM_A, STREAM_B);
+ context.start();
+
+ KinesisClient a = context.getEndpoint(STREAM_A,
Kinesis2Endpoint.class).getClient();
+ KinesisClient b = context.getEndpoint(STREAM_B,
Kinesis2Endpoint.class).getClient();
+ assertNotSame(a, b);
+ assertEquals("us-east-1",
a.serviceClientConfiguration().region().id());
+ assertEquals("eu-west-1",
b.serviceClientConfiguration().region().id());
+ }
+
+ @Test
+ public void eachEndpointCreatesTheAsyncClientOfItsOwnConfiguration()
throws Exception {
+ String a = STREAM_A + "&asyncClient=true";
+ String b = STREAM_B + "&asyncClient=true";
+ addRoutes(a, b);
+ context.start();
+
+ assertEquals("us-east-1",
+ context.getEndpoint(a,
Kinesis2Endpoint.class).getAsyncClient().serviceClientConfiguration().region()
+ .id());
+ assertEquals("eu-west-1",
+ context.getEndpoint(b,
Kinesis2Endpoint.class).getAsyncClient().serviceClientConfiguration().region()
+ .id());
+ }
+
+ @Test
+ public void producerUsesTheClientConfiguredOnItsEndpoint() throws
Exception {
+ KinesisClient clientA = mockClient();
+ KinesisClient clientB = mockClient();
+ context.getRegistry().bind("clientA", clientA);
+ context.getRegistry().bind("clientB", clientB);
+ addRoutes("aws2-kinesis:streamA?amazonKinesisClient=#clientA",
"aws2-kinesis:streamB?amazonKinesisClient=#clientB");
+ context.start();
+
+ send("direct:b");
+
+ Mockito.verify(clientB).putRecord(Mockito.any(PutRecordRequest.class));
+ Mockito.verify(clientA,
Mockito.never()).putRecord(Mockito.any(PutRecordRequest.class));
+ }
+
+ @Test
+ public void removingARouteKeepsTheClientOfAnotherEndpointOpen() throws
Exception {
+ addRoutes(STREAM_A, STREAM_A.replace("streamA", "streamB"));
+ context.start();
+
+ context.getRouteController().stopRoute("a");
+ context.removeRoute("a");
+
+ // still connects (and fails to connect), instead of failing with
"Connection pool shut down"
+ Exception cause = send("direct:b").getException();
+ assertInstanceOf(SdkClientException.class, cause, () -> "unexpected
failure: " + cause);
+ }
+
+ @Test
+ public void clientConfiguredOnTheComponentIsNotClosed() throws Exception {
+ KinesisClient client = mockClient();
+ context.getComponent("aws2-kinesis",
Kinesis2Component.class).getConfiguration().setAmazonKinesisClient(client);
+ addRoutes("aws2-kinesis:streamA", "aws2-kinesis:streamB");
+ context.start();
+ send("direct:a");
+
+ context.stop();
+
+ Mockito.verify(client).putRecord(Mockito.any(PutRecordRequest.class));
+ Mockito.verify(client, Mockito.never()).close();
+ }
+
+ private void addRoutes(String a, String b) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:a").routeId("a").to(a);
+ from("direct:b").routeId("b").to(b);
+ }
+ });
+ }
+
+ private Exchange send(String uri) {
+ return context.createProducerTemplate().send(uri, exchange -> {
+ exchange.getMessage().setHeader(Kinesis2Constants.PARTITION_KEY,
"key");
+ exchange.getMessage().setBody("record");
+ });
+ }
+
+ private static KinesisClient mockClient() {
+ KinesisClient client = Mockito.mock(KinesisClient.class);
+ Mockito.when(client.putRecord(Mockito.any(PutRecordRequest.class)))
+
.thenReturn(PutRecordResponse.builder().sequenceNumber("1").shardId("shardId-000000000000").build());
+ return client;
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 351dcbdf45de..3318a7a7ea2e 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -239,6 +239,16 @@ which embeds its own relocated copy of Vert.x. The `vertx`
option of `camel-aran
See the
https://github.com/fabric8io/kubernetes-client/blob/main/doc/MIGRATION-v8.md[Kubernetes
client 8 migration guide]
for the other changes.
+=== camel-aws2-kinesis - each endpoint uses its own client
+
+Each `aws2-kinesis` endpoint now uses the Kinesis client of its own
configuration (`region`, credentials,
+`uriEndpointOverride`, or `amazonKinesisClient` / `amazonKinesisAsyncClient`).
Before, all endpoints of the component
+used the client of the endpoint that started first, and stopping one endpoint
(for example by removing its route)
+closed that client for every other endpoint.
+
+Camel now closes only the clients it created. A client set with
`amazonKinesisClient` or `amazonKinesisAsyncClient`, on
+the component or on an endpoint, is no longer closed when Camel stops: close
it yourself when you no longer need it.
+
=== camel-gson, camel-jsonb - the unmarshal type header needs
allowUnmarshallType
The Gson and JSON-B data formats now use the `CamelGsonUnmarshalType` and
`CamelJsonbUnmarshallType` headers to choose
