shashank created CAMEL-25909:
--------------------------------
Summary: camel-aws2-kinesis - all endpoints share the client of
the first endpoint (its region and credentials), and stopping one endpoint
closes it for all the others
Key: CAMEL-25909
URL: https://issues.apache.org/jira/browse/CAMEL-25909
Project: Camel
Issue Type: Bug
Components: camel-aws2-kinesis
Reporter: shashank
Assignee: shashank
{{Kinesis2Component}} holds one {{KinesisConnection}} for all its endpoints
(main 727cab81a8b8, {{Kinesis2Component}} line 35).
{{KinesisConnection.getClient(endpoint)}} and {{getAsyncClient(endpoint)}}
build the client from the configuration of the endpoint that asks first, or
take its {{amazonKinesisClient}}, and then hand that same client to every other
aws2-kinesis endpoint. {{Kinesis2Endpoint.doStart}}, the producer
({{Kinesis2Producer}} lines 94 and 156) and the polling consumer all use it. So:
* an endpoint uses the region, credentials and {{uriEndpointOverride}} of
another endpoint.
{{aws2-kinesis:streamB?accessKey=b&secretKey=b®ion=eu-west-1}} next to
{{aws2-kinesis:streamA?accessKey=a&secretKey=a®ion=us-east-1}} reads and
writes {{streamB}} in us-east-1 with the credentials of A;
* a client given on the endpoint is ignored once another endpoint has started:
with {{amazonKinesisClient=#clientB}}, the records of route B are put through
{{#clientA}};
* {{Kinesis2Endpoint.doStop}} (lines 81-88) closes that shared client when the
endpoint has no {{amazonKinesisClient}}, and the connection keeps the closed
instance. After one aws2-kinesis endpoint stops, every other aws2-kinesis
producer and consumer of the context fails every call with
{{IllegalStateException: Connection pool shut down}} until the context
restarts. An endpoint stops when its route is removed ({{AbstractCamelContext}}
removes the endpoints that no other route uses), when routes are reloaded, when
a kamelet route is removed, or on {{removeEndpoint}};
* the component's {{doStop}} closes a client that the user set on the component.
CAMEL-24156 (item 4 of its camel-aws2-kinesis section) reported this. PR
#24863, which resolved that umbrella, only made {{KinesisConnection.close()}}
(component stop) clear its fields. The cache shared by the endpoints and the
close in the endpoint's {{doStop}} are unchanged.
*Fix*: each {{Kinesis2Endpoint}} has its own {{KinesisConnection}} and gives it
to its producer and consumer. The connection records which clients it created
and closes only those, never a client set with {{amazonKinesisClient}} /
{{amazonKinesisAsyncClient}} on the endpoint or the component. The endpoint's
{{doStop}} closes its connection, and a later start creates new clients.
{{Kinesis2Component.getConnection()}} is kept, because tests wire consumers
with it. Upgrade note in the 4.24 guide: Camel no longer closes a client that
the user set on the component or on an endpoint (before, the component closed
the cached one at stop).
*Tests*: new {{KinesisEndpointClientTest}} (5 tests, JUnit assertions and
Mockito as in the module), all fail on main:
* {{eachEndpointCreatesTheClientOfItsOwnConfiguration}}: {{expected: not same
but was: <DefaultKinesisClient@...>}};
* {{eachEndpointCreatesTheAsyncClientOfItsOwnConfiguration}}
({{asyncClient=true}}): {{expected: <eu-west-1> but was: <us-east-1>}};
* {{producerUsesTheClientConfiguredOnItsEndpoint}}: {{putRecord}} goes to the
client of the other endpoint;
* {{removingARouteKeepsTheClientOfAnotherEndpointOpen}}: {{expected:
<SdkClientException> but was: <IllegalStateException>}} ({{Connection pool shut
down}});
* {{clientConfiguredOnTheComponentIsNotClosed}}: {{close()}} of the user's
client is called at context stop.
camel-aws2-kinesis unit tests with the fix: 51 run, 0 failures.
*How it was found*: while writing a deterministic simulation test of the
polling consumer, by reading where it gets its client. The JUnit tests above
are the evidence.
*Affected*: main, camel-4.23.x, camel-4.22.x, camel-4.18.x and camel-4.14.x
(the same component-wide connection, and the same close in
{{Kinesis2Endpoint.doStop}}; read, not run).
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
