This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 2a8b165e51c8 CAMEL-25813: camel-pulsar - the producer continues the
sequence ids of the broker (#28203)
2a8b165e51c8 is described below
commit 2a8b165e51c8261e443791c0359a84531d127386
Author: allthingssecurity <[email protected]>
AuthorDate: Thu Oct 22 00:37:53 2026 +0530
CAMEL-25813: camel-pulsar - the producer continues the sequence ids of the
broker (#28203)
* CAMEL-25813: camel-pulsar - the producer continues the sequence ids of
the broker
The producer always passed initialSequenceId (default -1) to the Pulsar
client. The client continues after the last sequence id the broker has
for the producer name only when no initial sequence id is set, so every
start of a producer with a producerName began again at sequence id 0.
With deduplication enabled on the broker, the messages sent after a
restart (of the application or of the route) were then dropped as
duplicates while the send succeeded, until the sequence ids passed the
ones of the previous run.
The initial sequence id is now only passed when it is 0 or more.
* CAMEL-25813: camel-pulsar - address review: the comment on the initial
sequence id goes above its check
It also says that any negative value (the default -1) leaves the initial
sequence id to the Pulsar client, as the check is >= 0.
* CAMEL-25813: camel-pulsar - address review: test initialSequenceId=0, and
say in the doc that a set value starts every run there
From Andrea Cosentino's review of #28203:
- New test testAConfiguredInitialSequenceIdOfZeroIsUsed:
initialSequenceId=0 is passed to the client.
It fails if the check is "> 0" instead of ">= 0" (expected: <0> but was:
<null>).
- The component page (and its catalog copy) now says that with
initialSequenceId set every start
begins after that value again, so with a producerName and deduplication
the broker drops the
messages until they pass its last sequence id, and that leaving it unset
lets the client continue.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/components/pulsar.json | 4 +-
.../camel/catalog/docs/pulsar-component.adoc | 8 ++
.../org/apache/camel/component/pulsar/pulsar.json | 4 +-
.../src/main/docs/pulsar-component.adoc | 8 ++
.../component/pulsar/PulsarConfiguration.java | 10 +-
.../camel/component/pulsar/PulsarProducer.java | 8 +-
.../PulsarProducerInitialSequenceIdTest.java | 145 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 8 ++
.../dsl/PulsarComponentBuilderFactory.java | 4 +-
.../endpoint/dsl/PulsarEndpointBuilderFactory.java | 8 +-
10 files changed, 196 insertions(+), 11 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/pulsar.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/pulsar.json
index edd57b05efb8..14015d1ae1fa 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/pulsar.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/pulsar.json
@@ -63,7 +63,7 @@
"chunkingEnabled": { "index": 36, "kind": "property", "displayName":
"Chunking Enabled", "group": "producer", "label": "producer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false,
"configurationClass": "org.apache.camel.component.pulsar.PulsarConfiguration",
"configurationField": "configuration", "description": "Control whether chunking
of messages is enabled for the producer." },
"compressionType": { "index": 37, "kind": "property", "displayName":
"Compression Type", "group": "producer", "label": "producer", "required":
false, "type": "enum", "javaType":
"org.apache.pulsar.client.api.CompressionType", "enum": [ "NONE", "LZ4",
"ZLIB", "ZSTD", "SNAPPY" ], "deprecated": false, "autowired": false, "secret":
false, "defaultValue": "NONE", "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"configuration", "descript [...]
"hashingScheme": { "index": 38, "kind": "property", "displayName":
"Hashing Scheme", "group": "producer", "label": "producer", "required": false,
"type": "enum", "javaType": "java.lang.String", "enum": [ "JavaStringHash",
"Murmur3_32Hash" ], "deprecated": false, "autowired": false, "secret": false,
"defaultValue": "JavaStringHash", "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"configuration", "description": "Hashing function to [...]
- "initialSequenceId": { "index": 39, "kind": "property", "displayName":
"Initial Sequence Id", "group": "producer", "label": "producer", "required":
false, "type": "integer", "javaType": "long", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": -1, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"configuration", "description": "The first message published will have a
sequence Id of initialSequenceId 1." },
+ "initialSequenceId": { "index": 39, "kind": "property", "displayName":
"Initial Sequence Id", "group": "producer", "label": "producer", "required":
false, "type": "integer", "javaType": "long", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": -1, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"configuration", "description": "The first message published will have a
sequence Id of initialSequenceId plus one. [...]
"lazyStartProducer": { "index": 40, "kind": "property", "displayName":
"Lazy Start Producer", "group": "producer", "label": "producer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false, "description":
"Whether the producer should be started lazy (on the first message). By
starting lazy you can use this to allow CamelContext and routes to startup in
situations where a producer may otherwise fai [...]
"maxPendingMessages": { "index": 41, "kind": "property", "displayName":
"Max Pending Messages", "group": "producer", "label": "producer", "required":
false, "type": "integer", "javaType": "int", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": 1000, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"configuration", "description": "Size of the pending massages queue. When the
queue is full, by default, any fur [...]
"maxPendingMessagesAcrossPartitions": { "index": 42, "kind": "property",
"displayName": "Max Pending Messages Across Partitions", "group": "producer",
"label": "producer", "required": false, "type": "integer", "javaType": "int",
"deprecated": true, "autowired": false, "secret": false, "defaultValue": 50000,
"configurationClass": "org.apache.camel.component.pulsar.PulsarConfiguration",
"configurationField": "configuration", "description": "The maximum number of
pending messages for pa [...]
@@ -138,7 +138,7 @@
"chunkingEnabled": { "index": 40, "kind": "parameter", "displayName":
"Chunking Enabled", "group": "producer", "label": "producer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false,
"configurationClass": "org.apache.camel.component.pulsar.PulsarConfiguration",
"configurationField": "pulsarConfiguration", "description": "Control whether
chunking of messages is enabled for the producer." },
"compressionType": { "index": 41, "kind": "parameter", "displayName":
"Compression Type", "group": "producer", "label": "producer", "required":
false, "type": "enum", "javaType":
"org.apache.pulsar.client.api.CompressionType", "enum": [ "NONE", "LZ4",
"ZLIB", "ZSTD", "SNAPPY" ], "deprecated": false, "autowired": false, "secret":
false, "defaultValue": "NONE", "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"pulsarConfiguration", "d [...]
"hashingScheme": { "index": 42, "kind": "parameter", "displayName":
"Hashing Scheme", "group": "producer", "label": "producer", "required": false,
"type": "enum", "javaType": "java.lang.String", "enum": [ "JavaStringHash",
"Murmur3_32Hash" ], "deprecated": false, "autowired": false, "secret": false,
"defaultValue": "JavaStringHash", "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"pulsarConfiguration", "description": "Hashing funct [...]
- "initialSequenceId": { "index": 43, "kind": "parameter", "displayName":
"Initial Sequence Id", "group": "producer", "label": "producer", "required":
false, "type": "integer", "javaType": "long", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": -1, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"pulsarConfiguration", "description": "The first message published will have a
sequence Id of initialSequenceId 1." },
+ "initialSequenceId": { "index": 43, "kind": "parameter", "displayName":
"Initial Sequence Id", "group": "producer", "label": "producer", "required":
false, "type": "integer", "javaType": "long", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": -1, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"pulsarConfiguration", "description": "The first message published will have a
sequence Id of initialSequenceId pl [...]
"maxPendingMessages": { "index": 44, "kind": "parameter", "displayName":
"Max Pending Messages", "group": "producer", "label": "producer", "required":
false, "type": "integer", "javaType": "int", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": 1000, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"pulsarConfiguration", "description": "Size of the pending massages queue. When
the queue is full, by default, [...]
"maxPendingMessagesAcrossPartitions": { "index": 45, "kind": "parameter",
"displayName": "Max Pending Messages Across Partitions", "group": "producer",
"label": "producer", "required": false, "type": "integer", "javaType": "int",
"deprecated": true, "autowired": false, "secret": false, "defaultValue": 50000,
"configurationClass": "org.apache.camel.component.pulsar.PulsarConfiguration",
"configurationField": "pulsarConfiguration", "description": "The maximum number
of pending messages [...]
"messageRouter": { "index": 46, "kind": "parameter", "displayName":
"Message Router", "group": "producer", "label": "producer", "required": false,
"type": "object", "javaType": "org.apache.pulsar.client.api.MessageRouter",
"deprecated": false, "autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"pulsarConfiguration", "description": "Custom Message Router to use" },
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/pulsar-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/pulsar-component.adoc
index efa4a19b9725..dd055e1772b3 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/pulsar-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/pulsar-component.adoc
@@ -72,3 +72,11 @@ The following routes send a message to the `orders` topic
every five seconds, an
- log:
message: "Received ${body}"
----
+
+=== Producer sequence ids and deduplication
+
+With a `producerName` and deduplication enabled on the broker, the producer
continues the sequence ids after the
+last sequence id the broker stored for that producer name, also after a
restart, so the broker does not drop the
+messages of the new run as duplicates. Set `initialSequenceId` only to start
from a given sequence id: every start
+then begins after that value again, so with a `producerName` and deduplication
the broker drops the messages until
+their sequence ids pass the last one it stored. Leave the option unset to let
the client continue.
diff --git
a/components/camel-pulsar/src/generated/resources/META-INF/org/apache/camel/component/pulsar/pulsar.json
b/components/camel-pulsar/src/generated/resources/META-INF/org/apache/camel/component/pulsar/pulsar.json
index edd57b05efb8..14015d1ae1fa 100644
---
a/components/camel-pulsar/src/generated/resources/META-INF/org/apache/camel/component/pulsar/pulsar.json
+++
b/components/camel-pulsar/src/generated/resources/META-INF/org/apache/camel/component/pulsar/pulsar.json
@@ -63,7 +63,7 @@
"chunkingEnabled": { "index": 36, "kind": "property", "displayName":
"Chunking Enabled", "group": "producer", "label": "producer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false,
"configurationClass": "org.apache.camel.component.pulsar.PulsarConfiguration",
"configurationField": "configuration", "description": "Control whether chunking
of messages is enabled for the producer." },
"compressionType": { "index": 37, "kind": "property", "displayName":
"Compression Type", "group": "producer", "label": "producer", "required":
false, "type": "enum", "javaType":
"org.apache.pulsar.client.api.CompressionType", "enum": [ "NONE", "LZ4",
"ZLIB", "ZSTD", "SNAPPY" ], "deprecated": false, "autowired": false, "secret":
false, "defaultValue": "NONE", "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"configuration", "descript [...]
"hashingScheme": { "index": 38, "kind": "property", "displayName":
"Hashing Scheme", "group": "producer", "label": "producer", "required": false,
"type": "enum", "javaType": "java.lang.String", "enum": [ "JavaStringHash",
"Murmur3_32Hash" ], "deprecated": false, "autowired": false, "secret": false,
"defaultValue": "JavaStringHash", "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"configuration", "description": "Hashing function to [...]
- "initialSequenceId": { "index": 39, "kind": "property", "displayName":
"Initial Sequence Id", "group": "producer", "label": "producer", "required":
false, "type": "integer", "javaType": "long", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": -1, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"configuration", "description": "The first message published will have a
sequence Id of initialSequenceId 1." },
+ "initialSequenceId": { "index": 39, "kind": "property", "displayName":
"Initial Sequence Id", "group": "producer", "label": "producer", "required":
false, "type": "integer", "javaType": "long", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": -1, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"configuration", "description": "The first message published will have a
sequence Id of initialSequenceId plus one. [...]
"lazyStartProducer": { "index": 40, "kind": "property", "displayName":
"Lazy Start Producer", "group": "producer", "label": "producer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false, "description":
"Whether the producer should be started lazy (on the first message). By
starting lazy you can use this to allow CamelContext and routes to startup in
situations where a producer may otherwise fai [...]
"maxPendingMessages": { "index": 41, "kind": "property", "displayName":
"Max Pending Messages", "group": "producer", "label": "producer", "required":
false, "type": "integer", "javaType": "int", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": 1000, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"configuration", "description": "Size of the pending massages queue. When the
queue is full, by default, any fur [...]
"maxPendingMessagesAcrossPartitions": { "index": 42, "kind": "property",
"displayName": "Max Pending Messages Across Partitions", "group": "producer",
"label": "producer", "required": false, "type": "integer", "javaType": "int",
"deprecated": true, "autowired": false, "secret": false, "defaultValue": 50000,
"configurationClass": "org.apache.camel.component.pulsar.PulsarConfiguration",
"configurationField": "configuration", "description": "The maximum number of
pending messages for pa [...]
@@ -138,7 +138,7 @@
"chunkingEnabled": { "index": 40, "kind": "parameter", "displayName":
"Chunking Enabled", "group": "producer", "label": "producer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false,
"configurationClass": "org.apache.camel.component.pulsar.PulsarConfiguration",
"configurationField": "pulsarConfiguration", "description": "Control whether
chunking of messages is enabled for the producer." },
"compressionType": { "index": 41, "kind": "parameter", "displayName":
"Compression Type", "group": "producer", "label": "producer", "required":
false, "type": "enum", "javaType":
"org.apache.pulsar.client.api.CompressionType", "enum": [ "NONE", "LZ4",
"ZLIB", "ZSTD", "SNAPPY" ], "deprecated": false, "autowired": false, "secret":
false, "defaultValue": "NONE", "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"pulsarConfiguration", "d [...]
"hashingScheme": { "index": 42, "kind": "parameter", "displayName":
"Hashing Scheme", "group": "producer", "label": "producer", "required": false,
"type": "enum", "javaType": "java.lang.String", "enum": [ "JavaStringHash",
"Murmur3_32Hash" ], "deprecated": false, "autowired": false, "secret": false,
"defaultValue": "JavaStringHash", "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"pulsarConfiguration", "description": "Hashing funct [...]
- "initialSequenceId": { "index": 43, "kind": "parameter", "displayName":
"Initial Sequence Id", "group": "producer", "label": "producer", "required":
false, "type": "integer", "javaType": "long", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": -1, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"pulsarConfiguration", "description": "The first message published will have a
sequence Id of initialSequenceId 1." },
+ "initialSequenceId": { "index": 43, "kind": "parameter", "displayName":
"Initial Sequence Id", "group": "producer", "label": "producer", "required":
false, "type": "integer", "javaType": "long", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": -1, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"pulsarConfiguration", "description": "The first message published will have a
sequence Id of initialSequenceId pl [...]
"maxPendingMessages": { "index": 44, "kind": "parameter", "displayName":
"Max Pending Messages", "group": "producer", "label": "producer", "required":
false, "type": "integer", "javaType": "int", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": 1000, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"pulsarConfiguration", "description": "Size of the pending massages queue. When
the queue is full, by default, [...]
"maxPendingMessagesAcrossPartitions": { "index": 45, "kind": "parameter",
"displayName": "Max Pending Messages Across Partitions", "group": "producer",
"label": "producer", "required": false, "type": "integer", "javaType": "int",
"deprecated": true, "autowired": false, "secret": false, "defaultValue": 50000,
"configurationClass": "org.apache.camel.component.pulsar.PulsarConfiguration",
"configurationField": "pulsarConfiguration", "description": "The maximum number
of pending messages [...]
"messageRouter": { "index": 46, "kind": "parameter", "displayName":
"Message Router", "group": "producer", "label": "producer", "required": false,
"type": "object", "javaType": "org.apache.pulsar.client.api.MessageRouter",
"deprecated": false, "autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"pulsarConfiguration", "description": "Custom Message Router to use" },
diff --git a/components/camel-pulsar/src/main/docs/pulsar-component.adoc
b/components/camel-pulsar/src/main/docs/pulsar-component.adoc
index efa4a19b9725..dd055e1772b3 100644
--- a/components/camel-pulsar/src/main/docs/pulsar-component.adoc
+++ b/components/camel-pulsar/src/main/docs/pulsar-component.adoc
@@ -72,3 +72,11 @@ The following routes send a message to the `orders` topic
every five seconds, an
- log:
message: "Received ${body}"
----
+
+=== Producer sequence ids and deduplication
+
+With a `producerName` and deduplication enabled on the broker, the producer
continues the sequence ids after the
+last sequence id the broker stored for that producer name, also after a
restart, so the broker does not drop the
+messages of the new run as duplicates. Set `initialSequenceId` only to start
from a given sequence id: every start
+then begins after that value again, so with a `producerName` and deduplication
the broker drops the messages until
+their sequence ids pass the last one it stored. Leave the option unset to let
the client continue.
diff --git
a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/PulsarConfiguration.java
b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/PulsarConfiguration.java
index 0930f127e37c..12c2926be563 100644
---
a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/PulsarConfiguration.java
+++
b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/PulsarConfiguration.java
@@ -137,7 +137,10 @@ public class PulsarConfiguration implements Cloneable {
@UriParam(label = "producer", description = "Control batching method used
by the producer.", defaultValue = "DEFAULT",
enums = "DEFAULT,KEY_BASED")
private BatcherBuilder batcherBuilder = BatcherBuilder.DEFAULT;
- @UriParam(label = "producer", description = "The first message published
will have a sequence Id of initialSequenceId 1.",
+ @UriParam(label = "producer",
+ description = "The first message published will have a sequence
Id of initialSequenceId plus one. The"
+ + " default -1 leaves it to the Pulsar client,
which continues after the last sequence id the"
+ + " broker has for the producer name.",
defaultValue = "-1")
private long initialSequenceId = -1;
@UriParam(label = "producer", description = "Compression type to use",
defaultValue = "NONE",
@@ -454,8 +457,9 @@ public class PulsarConfiguration implements Cloneable {
/**
* Set the baseline for the sequence ids for messages published by the
producer. First message will be using
- * (initialSequenceId 1) as its sequence id and subsequent messages will
be assigned incremental sequence ids, if
- * not otherwise specified.
+ * (initialSequenceId + 1) as its sequence id and subsequent messages will
be assigned incremental sequence ids, if
+ * not otherwise specified. The default -1 leaves it to the Pulsar client,
which continues after the last sequence
+ * id the broker has for the producer name (needed for broker
deduplication across restarts).
*/
public void setInitialSequenceId(long initialSequenceId) {
this.initialSequenceId = initialSequenceId;
diff --git
a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/PulsarProducer.java
b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/PulsarProducer.java
index 88840855aaf6..7a40d2617d11 100644
---
a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/PulsarProducer.java
+++
b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/PulsarProducer.java
@@ -138,7 +138,6 @@ public class PulsarProducer extends DefaultAsyncProducer {
.batchingMaxPublishDelay(configuration.getBatchingMaxPublishDelayMicros(),
TimeUnit.MICROSECONDS)
.batchingMaxMessages(configuration.getMaxPendingMessages())
.enableBatching(configuration.isBatchingEnabled()).batcherBuilder(configuration.getBatcherBuilder())
-
.initialSequenceId(configuration.getInitialSequenceId())
.compressionType(configuration.getCompressionType())
.hashingScheme(HashingScheme.valueOf(configuration.getHashingScheme()))
.enableChunking(configuration.isChunkingEnabled());
@@ -150,6 +149,13 @@ public class PulsarProducer extends DefaultAsyncProducer {
if (producerName != null) {
producerBuilder.producerName(producerName);
}
+ // with a negative value (the default -1) the initial sequence
id is not set, so the Pulsar client
+ // continues after the last sequence id the broker has for the
producer name. Passing -1 would make
+ // every start begin again at sequence id 0, and with
deduplication enabled the broker would drop the
+ // messages of the new run as duplicates of the previous one
+ if (configuration.getInitialSequenceId() >= 0) {
+
producerBuilder.initialSequenceId(configuration.getInitialSequenceId());
+ }
producer = producerBuilder.create();
}
} finally {
diff --git
a/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/PulsarProducerInitialSequenceIdTest.java
b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/PulsarProducerInitialSequenceIdTest.java
new file mode 100644
index 000000000000..5722db77b5e5
--- /dev/null
+++
b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/PulsarProducerInitialSequenceIdTest.java
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.pulsar;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.TypedMessageBuilder;
+import org.apache.pulsar.client.impl.ProducerBuilderImpl;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.Answers.RETURNS_SELF;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+/**
+ * The Pulsar client continues the sequence ids of a named producer after the
last sequence id the broker stored for it
+ * only when no initial sequence id is configured. A producer that always
passed one (-1 by default) started again at
+ * sequence id 0 after every start, and with deduplication enabled the broker
dropped the messages as duplicates until
+ * the sequence ids passed the ones of the previous run.
+ */
+public class PulsarProducerInitialSequenceIdTest extends CamelTestSupport {
+
+ private static final String NAMED_TOPIC =
"persistent://public/default/camel-sequence-test";
+ private static final String CONFIGURED_TOPIC =
"persistent://public/default/camel-sequence-test-2";
+ private static final String ZERO_TOPIC =
"persistent://public/default/camel-sequence-test-3";
+
+ private final List<ProducerBuilderImpl<byte[]>> builders = new
ArrayList<>();
+
+ private List<ProducerBuilderImpl<byte[]>> builders(String topic) {
+ return builders.stream().filter(builder ->
topic.equals(builder.getConf().getTopicName())).toList();
+ }
+
+ @Test
+ public void testTheBrokerSequenceIdIsUsedByDefault() {
+ template.sendBody("direct:named", "Hello");
+
+ List<ProducerBuilderImpl<byte[]>> named = builders(NAMED_TOPIC);
+ assertEquals(1, named.size());
+ assertNull(named.get(0).getConf().getInitialSequenceId(),
+ "no initial sequence id must be set, so that the client
continues after the last sequence id of the broker");
+ }
+
+ @Test
+ public void testTheBrokerSequenceIdIsUsedAfterARestart() throws Exception {
+ template.sendBody("direct:named", "Hello");
+ context.getRouteController().stopRoute("named");
+ context.getRouteController().startRoute("named");
+ template.sendBody("direct:named", "Hello again");
+
+ List<ProducerBuilderImpl<byte[]>> named = builders(NAMED_TOPIC);
+ assertEquals(2, named.size(), "the producer is created again after the
restart");
+ assertNull(named.get(1).getConf().getInitialSequenceId(),
+ "no initial sequence id must be set, so that the client
continues after the last sequence id of the broker");
+ }
+
+ @Test
+ public void testAConfiguredInitialSequenceIdIsUsed() {
+ template.sendBody("direct:configured", "Hello");
+
+ List<ProducerBuilderImpl<byte[]>> configured =
builders(CONFIGURED_TOPIC);
+ assertEquals(1, configured.size());
+ assertEquals(41L, configured.get(0).getConf().getInitialSequenceId());
+ }
+
+ @Test
+ public void testAConfiguredInitialSequenceIdOfZeroIsUsed() {
+ template.sendBody("direct:zero", "Hello");
+
+ List<ProducerBuilderImpl<byte[]>> zero = builders(ZERO_TOPIC);
+ assertEquals(1, zero.size());
+ assertEquals(0L, zero.get(0).getConf().getInitialSequenceId());
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ final CamelContext context = super.createCamelContext();
+
+ final TypedMessageBuilder<byte[]> messageBuilder =
mock(TypedMessageBuilder.class, RETURNS_SELF);
+
when(messageBuilder.sendAsync()).thenReturn(CompletableFuture.completedFuture(mock(MessageId.class)));
+
+ final Producer<byte[]> pulsarProducer = mock(Producer.class);
+ when(pulsarProducer.newMessage()).thenReturn(messageBuilder);
+
+ final PulsarClient pulsarClient = mock(PulsarClient.class);
+ when(pulsarClient.newProducer()).thenAnswer(invocation -> {
+ // the real builder of the Pulsar client, which keeps the
configuration the producer is created with
+ final ProducerBuilderImpl<byte[]> builder = spy(new
ProducerBuilderImpl<>(null, Schema.BYTES));
+ doReturn(pulsarProducer).when(builder).create();
+ builders.add(builder);
+ return builder;
+ });
+
+ final PulsarComponent component = new PulsarComponent(context);
+ component.setPulsarClient(pulsarClient);
+ context.addComponent("pulsar", component);
+
+ return context;
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:named").routeId("named")
+ .to("pulsar:" + NAMED_TOPIC +
"?producerName=camel-producer");
+
+ from("direct:configured").routeId("configured")
+ .to("pulsar:" + CONFIGURED_TOPIC +
"?producerName=camel-producer-2"
+ + "&initialSequenceId=41");
+
+ from("direct:zero").routeId("zero")
+ .to("pulsar:" + ZERO_TOPIC +
"?producerName=camel-producer-3&initialSequenceId=0");
+ }
+ };
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index db8c88aa6b90..bcae0bed3e0b 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -1490,6 +1490,14 @@ live, so it never expired, even when the endpoint has a
`timeToLive` and, for re
`requestTimeout` as time to live otherwise. Such a message now gets the time
to live of the endpoint, as the option
documents. Set `disableTimeToLive=true` to send it without time to live.
+=== camel-pulsar - the producer continues the sequence ids of the broker
+
+The `pulsar` producer no longer passes the default `initialSequenceId=-1` to
the Pulsar client, so the client
+continues after the last sequence id the broker has for the `producerName`.
Before, every start began again at
+sequence id 0, and with deduplication enabled on the broker the messages sent
after a restart were dropped as
+duplicates (the send succeeded) until the sequence ids passed the ones of the
previous run. A configured
+`initialSequenceId` of 0 or more is still used as before.
+
=== camel-twitter - twitter-search applies its query options again
Since Camel 4.0 (the upgrade to Twitter4J 4.1, whose `Query` is immutable) the
`twitter-search` endpoint sent only the
diff --git
a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/PulsarComponentBuilderFactory.java
b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/PulsarComponentBuilderFactory.java
index 7c019758db60..3729ba62f5ef 100644
---
a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/PulsarComponentBuilderFactory.java
+++
b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/PulsarComponentBuilderFactory.java
@@ -752,7 +752,9 @@ public interface PulsarComponentBuilderFactory {
/**
* The first message published will have a sequence Id of
- * initialSequenceId 1.
+ * initialSequenceId plus one. The default -1 leaves it to the Pulsar
+ * client, which continues after the last sequence id the broker has
for
+ * the producer name.
*
* The option is a: <code>long</code> type.
*
diff --git
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/PulsarEndpointBuilderFactory.java
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/PulsarEndpointBuilderFactory.java
index 3a73d95d10c0..70b5d2557d17 100644
---
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/PulsarEndpointBuilderFactory.java
+++
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/PulsarEndpointBuilderFactory.java
@@ -1223,7 +1223,9 @@ public interface PulsarEndpointBuilderFactory {
}
/**
* The first message published will have a sequence Id of
- * initialSequenceId 1.
+ * initialSequenceId plus one. The default -1 leaves it to the Pulsar
+ * client, which continues after the last sequence id the broker has
for
+ * the producer name.
*
* The option is a: <code>long</code> type.
*
@@ -1239,7 +1241,9 @@ public interface PulsarEndpointBuilderFactory {
}
/**
* The first message published will have a sequence Id of
- * initialSequenceId 1.
+ * initialSequenceId plus one. The default -1 leaves it to the Pulsar
+ * client, which continues after the last sequence id the broker has
for
+ * the producer name.
*
* The option will be converted to a <code>long</code> type.
*
