shashank created CAMEL-25813:
--------------------------------
Summary: camel-pulsar - with a producerName and broker
deduplication, the producer restarts at sequence id 0 and the broker drops the
messages after every restart
Key: CAMEL-25813
URL: https://issues.apache.org/jira/browse/CAMEL-25813
Project: Camel
Issue Type: Bug
Components: camel-pulsar
Reporter: shashank
Assignee: shashank
{{PulsarProducer.createProducer}} (main 62c9a3ebe409, line 141) always passes
the {{initialSequenceId}} option to the Pulsar client builder:
{{.initialSequenceId(configuration.getInitialSequenceId())}}, and the option
defaults to -1 ({{PulsarConfiguration}} line 142).
The Pulsar client (pulsar-client 4.2.4, {{ProducerImpl}}; read from the
bytecode) uses the last sequence id the broker returns for the producer name
when the producer connects only if {{conf.getInitialSequenceId() == null}} (in
{{connectionOpened}}: {{if (msgIdGenerator == 0 && conf.getInitialSequenceId()
== null) { lastSequenceIdPublished = lastSequenceId; msgIdGenerator =
lastSequenceId + 1; }}}). With -1 passed explicitly, every new producer starts
at sequence id 0.
Broker deduplication ({{brokerDeduplicationEnabled}}, or the namespace / topic
deduplication policy) needs a stable {{producerName}} and drops every message
whose sequence id is not above the last one stored for that producer name. So
with a {{producerName}} set and deduplication enabled (the setup Pulsar
documents for deduplication), after every start of the Camel producer
(application restart, route stop/start, reload) the broker drops the messages
as duplicates until the sequence ids pass the ones of the previous run: after a
run that sent 10,000 messages, the next 10,000 messages are lost. The broker
answers a dropped duplicate with a normal send receipt (ledger and entry id
-1), so the send completes, the exchange succeeds and nothing is logged (read,
not run). Without deduplication, or without a {{producerName}}, nothing changes.
*Fix*: pass {{initialSequenceId}} to the builder only when it is 0 or more; the
default -1 leaves it unset, so the client continues after the broker's last
sequence id for the producer name. Javadoc of the option, a short section in
the component page (and its catalog copy), upgrade note in the 4.24 guide. No
option or default changes, so no generated metadata changes.
*Tests*: new {{PulsarProducerInitialSequenceIdTest}} (a mocked {{PulsarClient}}
that returns the real {{ProducerBuilderImpl}} of the Pulsar client, with only
{{create()}} stubbed, so the test reads the configuration the producer would be
created with). {{testTheBrokerSequenceIdIsUsedByDefault}} and
{{testTheBrokerSequenceIdIsUsedAfterARestart}} (route stopped and started) fail
on main: {{no initial sequence id must be set, so that the client continues
after the last sequence id of the broker ==> expected: <null> but was: <-1>}}.
{{testAConfiguredInitialSequenceIdIsUsed}} ({{initialSequenceId=41}}) passes on
main and with the fix. camel-pulsar unit tests: 33 run, 0 failures (the ITs
need Docker and were not run).
*How it was found*: variant analysis of CAMEL-25789 (camel-aws2-s3 stream
upload, pattern "unique names or ids built from a counter that restarts at
every start, so a restart collides with what the previous run wrote"), then
reading how the Pulsar client initialises its sequence ids, then the JUnit test
above.
*Affected*: main, camel-4.23.x, camel-4.22.x, camel-4.18.x and camel-4.14.x
(the same unconditional {{initialSequenceId}} call and default -1; read, not
run). The drop by the broker was not run against a Pulsar broker here; it is
the documented deduplication behaviour and follows from the client code above.
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
