[
https://issues.apache.org/jira/browse/CAMEL-25813?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen resolved CAMEL-25813.
---------------------------------
Resolution: Fixed
> 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
> Priority: Major
> Fix For: 4.24.0
>
>
> {{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)
