[jira] [Work started] (CAMEL-25813) camel-pulsar - with a producerName and broker deduplication, the producer restarts at sequence id 0 and the broker drops the messages after every restart

["shashank (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/CAMEL-25813?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Work on CAMEL-25813 started by shashank.
----------------------------------------
> 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
>
> {{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)