Federico Mariani created CAMEL-26647:
----------------------------------------
Summary: 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.23.0, 4.22.2
Reporter: Federico Mariani
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)
