[
https://issues.apache.org/jira/browse/CAMEL-26647?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen updated CAMEL-26647:
--------------------------------
Fix Version/s: 4.22.3
4.18.6
> camel-google-pubsub - manual ack of an aggregated exchange (synchronousPull)
> acks none of the grouped messages
> --------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-26647
> URL: https://issues.apache.org/jira/browse/CAMEL-26647
> Project: Camel
> Issue Type: Bug
> Components: camel-google-pubsub
> Affects Versions: 4.18.5, 4.22.2, 4.23.0
> Reporter: Federico Mariani
> Assignee: Federico Mariani
> Priority: Major
> Fix For: 4.18.6, 4.22.3, 4.24.0
>
>
> h3. Problem
> With {{synchronousPull=true}} and {{ackMode=NONE}} the consumer puts an
> {{AcknowledgeSync}} in the header
> {{GooglePubsubConstants.GOOGLE_PUBSUB_ACKNOWLEDGE}}
> ({{components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumer.java}}
> lines 415-421 on main, commit 7978db3d42dd). When the route acks (or nacks)
> an aggregated exchange whose body is a {{List<Exchange>}},
> {{AcknowledgeSync.getAckIdList}} is meant to collect the ack id of every
> grouped exchange. Inside the loop it reads the header of the *outer* exchange
> instead
> ({{components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/consumer/AcknowledgeSync.java}}
> line 75):
> {code:java}
> for (Object body : exchange.getIn().getBody(List.class)) {
> if (body instanceof Exchange) {
> String ackId =
> exchange.getIn().getHeader(GooglePubsubConstants.ACK_ID, String.class); //
> should be the element's
> {code}
> {{GroupedExchangeAggregationStrategy}} builds the outer exchange with {{new
> DefaultExchange(newExchange)}}, which has no {{CamelGooglePubsubMsgAckId}}
> header. So the request carries no ack id, none of the grouped messages is
> acked, and Pub/Sub delivers them again once their ack deadline expires. This
> repeats on every delivery. If a strategy keeps the first exchange as the
> holder, its id is sent N+1 times and the other ids never (read from the
> source, not run). {{nack}} has the same defect.
> h3. How to reproduce
> Measured with the Pub/Sub emulator from {{camel-test-infra-google-pubsub}}: a
> subscription with a 5 s ack deadline, 3 messages published, and this route:
> {code:java}
> from("google-pubsub:test-project:sub?synchronousPull=true&ackMode=NONE&maxMessagesPerPoll=10&maxDeliveryAttempts=0")
> .aggregate(constant(true), new
> GroupedExchangeAggregationStrategy()).completionSize(3)
> .process(e -> {
> List<Exchange> grouped = e.getIn().getBody(List.class);
>
> grouped.get(0).getIn().getHeader(GooglePubsubConstants.GOOGLE_PUBSUB_ACKNOWLEDGE,
> GooglePubsubAcknowledge.class).ack(e);
> });
> {code}
> A delegating {{SubscriberStub}} (through an overridden
> {{GooglePubsubComponent.getSubscriberStub}}) records each
> {{AcknowledgeRequest}}.
> ||Case||AcknowledgeRequest(s) sent||Redelivered after the deadline||
> |{{ack(aggregated)}}|one request with an empty id list (the emulator accepted
> it without error)|yes: the same 3 messages delivered again after the 5 s
> deadline|
> |control: {{ack(g)}} for each grouped exchange|3 requests with 1 id each|no,
> none in the 15 s observed|
> On the aggregated exchange the {{CamelGooglePubsubMsgAckId}} header was
> {{null}}, while the 3 grouped exchanges had their own ids. Not checked
> against the real service, which may reject an empty {{ack_ids}} with
> INVALID_ARGUMENT. In that case {{ack()}} throws and the result is the same
> redelivery.
> h3. Proposed change
> * In the loop, read {{ACK_ID}} from {{((Exchange) body).getIn()}}. Keep the
> outer exchange's own id when present, and remove duplicates.
> * Do not send an empty {{AcknowledgeRequest}} / {{ModifyAckDeadlineRequest}}:
> log a warning instead (or fail the call).
> * Tests: a unit test on {{AcknowledgeSync}} with a stubbed {{SubscriberStub}}
> supplier that checks the ids for a grouped body, for both ack and nack. Also
> an emulator IT with an aggregated manual ack that checks nothing is
> redelivered.
> * Docs: the "Manual Acknowledgement" section does not mention aggregated
> exchanges. It should state that acking a grouped exchange acks every grouped
> message with the synchronous consumer only. {{AcknowledgeAsync}} (the default
> asynchronous consumer) ignores the exchange argument and acks only the
> message its header came from.
> h3. Impact / upgrade
> * No default changes, so no upgrade-guide entry is needed.
> * Only manual ack ({{ackMode=NONE}}) of an exchange with a {{List<Exchange>}}
> body, with {{synchronousPull=true}}.
> * With the default {{ackMode=AUTO}}, each pulled message is acked by its own
> exchange's completion, which is not affected.
> * The same code is on {{camel-4.18.x}} and {{camel-4.22.x}} (same line 75)
> and in the 4.18.5, 4.22.2 and 4.23.0 releases, so the fix is a backport
> candidate.
> h3. Related
> * CAMEL-10178 (07f57c33962c): the original code read the ack id of each
> grouped exchange ({{ex.getIn()}} over the {{GROUPED_EXCHANGE}} list).
> * CAMEL-14514 (5ef63390a07a): rewrote it to iterate the List body and
> introduced this bug.
> * CAMEL-15252 (cc75c3cb9d9c): put the acknowledge object in a header for
> {{ackMode=NONE}}.
> * CAMEL-18447 (1a5463a8d31f): the stub supplier used by {{AcknowledgeSync}}.
> * CAMEL-21929: a different aggregation issue, on the producer side.
> _Claude Code on behalf of Federico Mariani_
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
