This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch hashingscheme-pulsar
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 073833ad142b6755e18e12ad9110c78b09cc773a
Author: Andrea Cosentino <anco...@gmail.com>
AuthorDate: Thu Sep 14 07:44:36 2023 +0200
CAMEL-19866 - Camel-Pulsar: Support HashingScheme as uri parameter for
producers
Signed-off-by: Andrea Cosentino <anco...@gmail.com>
---
.../pulsar/PulsarComponentConfigurer.java | 6 ++++
.../component/pulsar/PulsarEndpointConfigurer.java | 6 ++++
.../component/pulsar/PulsarEndpointUriFactory.java | 3 +-
.../org/apache/camel/component/pulsar/pulsar.json | 40 ++++++++++++----------
.../component/pulsar/PulsarConfiguration.java | 15 ++++++++
.../camel/component/pulsar/PulsarProducer.java | 6 ++--
6 files changed, 52 insertions(+), 24 deletions(-)
diff --git
a/components/camel-pulsar/src/generated/java/org/apache/camel/component/pulsar/PulsarComponentConfigurer.java
b/components/camel-pulsar/src/generated/java/org/apache/camel/component/pulsar/PulsarComponentConfigurer.java
index 401e0589ca8..129bd7006f7 100644
---
a/components/camel-pulsar/src/generated/java/org/apache/camel/component/pulsar/PulsarComponentConfigurer.java
+++
b/components/camel-pulsar/src/generated/java/org/apache/camel/component/pulsar/PulsarComponentConfigurer.java
@@ -71,6 +71,8 @@ public class PulsarComponentConfigurer extends
PropertyConfigurerSupport impleme
case "deadLetterTopic":
getOrCreateConfiguration(target).setDeadLetterTopic(property(camelContext,
java.lang.String.class, value)); return true;
case "enableretry":
case "enableRetry":
getOrCreateConfiguration(target).setEnableRetry(property(camelContext,
boolean.class, value)); return true;
+ case "hashingscheme":
+ case "hashingScheme":
getOrCreateConfiguration(target).setHashingScheme(property(camelContext,
java.lang.String.class, value)); return true;
case "initialsequenceid":
case "initialSequenceId":
getOrCreateConfiguration(target).setInitialSequenceId(property(camelContext,
long.class, value)); return true;
case "keysharedpolicy":
@@ -176,6 +178,8 @@ public class PulsarComponentConfigurer extends
PropertyConfigurerSupport impleme
case "deadLetterTopic": return java.lang.String.class;
case "enableretry":
case "enableRetry": return boolean.class;
+ case "hashingscheme":
+ case "hashingScheme": return java.lang.String.class;
case "initialsequenceid":
case "initialSequenceId": return long.class;
case "keysharedpolicy":
@@ -277,6 +281,8 @@ public class PulsarComponentConfigurer extends
PropertyConfigurerSupport impleme
case "deadLetterTopic": return
getOrCreateConfiguration(target).getDeadLetterTopic();
case "enableretry":
case "enableRetry": return
getOrCreateConfiguration(target).isEnableRetry();
+ case "hashingscheme":
+ case "hashingScheme": return
getOrCreateConfiguration(target).getHashingScheme();
case "initialsequenceid":
case "initialSequenceId": return
getOrCreateConfiguration(target).getInitialSequenceId();
case "keysharedpolicy":
diff --git
a/components/camel-pulsar/src/generated/java/org/apache/camel/component/pulsar/PulsarEndpointConfigurer.java
b/components/camel-pulsar/src/generated/java/org/apache/camel/component/pulsar/PulsarEndpointConfigurer.java
index 0e4231feb2f..20b74a871d0 100644
---
a/components/camel-pulsar/src/generated/java/org/apache/camel/component/pulsar/PulsarEndpointConfigurer.java
+++
b/components/camel-pulsar/src/generated/java/org/apache/camel/component/pulsar/PulsarEndpointConfigurer.java
@@ -63,6 +63,8 @@ public class PulsarEndpointConfigurer extends
PropertyConfigurerSupport implemen
case "exceptionHandler":
target.setExceptionHandler(property(camelContext,
org.apache.camel.spi.ExceptionHandler.class, value)); return true;
case "exchangepattern":
case "exchangePattern":
target.setExchangePattern(property(camelContext,
org.apache.camel.ExchangePattern.class, value)); return true;
+ case "hashingscheme":
+ case "hashingScheme":
target.getPulsarConfiguration().setHashingScheme(property(camelContext,
java.lang.String.class, value)); return true;
case "initialsequenceid":
case "initialSequenceId":
target.getPulsarConfiguration().setInitialSequenceId(property(camelContext,
long.class, value)); return true;
case "keysharedpolicy":
@@ -158,6 +160,8 @@ public class PulsarEndpointConfigurer extends
PropertyConfigurerSupport implemen
case "exceptionHandler": return
org.apache.camel.spi.ExceptionHandler.class;
case "exchangepattern":
case "exchangePattern": return org.apache.camel.ExchangePattern.class;
+ case "hashingscheme":
+ case "hashingScheme": return java.lang.String.class;
case "initialsequenceid":
case "initialSequenceId": return long.class;
case "keysharedpolicy":
@@ -254,6 +258,8 @@ public class PulsarEndpointConfigurer extends
PropertyConfigurerSupport implemen
case "exceptionHandler": return target.getExceptionHandler();
case "exchangepattern":
case "exchangePattern": return target.getExchangePattern();
+ case "hashingscheme":
+ case "hashingScheme": return
target.getPulsarConfiguration().getHashingScheme();
case "initialsequenceid":
case "initialSequenceId": return
target.getPulsarConfiguration().getInitialSequenceId();
case "keysharedpolicy":
diff --git
a/components/camel-pulsar/src/generated/java/org/apache/camel/component/pulsar/PulsarEndpointUriFactory.java
b/components/camel-pulsar/src/generated/java/org/apache/camel/component/pulsar/PulsarEndpointUriFactory.java
index 7c15d4ab5aa..557e8c44cbb 100644
---
a/components/camel-pulsar/src/generated/java/org/apache/camel/component/pulsar/PulsarEndpointUriFactory.java
+++
b/components/camel-pulsar/src/generated/java/org/apache/camel/component/pulsar/PulsarEndpointUriFactory.java
@@ -21,7 +21,7 @@ public class PulsarEndpointUriFactory extends
org.apache.camel.support.component
private static final Set<String> SECRET_PROPERTY_NAMES;
private static final Set<String> MULTI_VALUE_PREFIXES;
static {
- Set<String> props = new HashSet<>(48);
+ Set<String> props = new HashSet<>(49);
props.add("ackGroupTimeMillis");
props.add("ackTimeoutMillis");
props.add("ackTimeoutRedeliveryBackoff");
@@ -43,6 +43,7 @@ public class PulsarEndpointUriFactory extends
org.apache.camel.support.component
props.add("enableRetry");
props.add("exceptionHandler");
props.add("exchangePattern");
+ props.add("hashingScheme");
props.add("initialSequenceId");
props.add("keySharedPolicy");
props.add("lazyStartProducer");
diff --git
a/components/camel-pulsar/src/generated/resources/org/apache/camel/component/pulsar/pulsar.json
b/components/camel-pulsar/src/generated/resources/org/apache/camel/component/pulsar/pulsar.json
index 6c5f40eeefb..83dfd3de283 100644
---
a/components/camel-pulsar/src/generated/resources/org/apache/camel/component/pulsar/pulsar.json
+++
b/components/camel-pulsar/src/generated/resources/org/apache/camel/component/pulsar/pulsar.json
@@ -58,17 +58,18 @@
"blockIfQueueFull": { "index": 33, "kind": "property", "displayName":
"Block If Queue Full", "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": "Whether to block the
producing thread if pending messages queue is full or to [...]
"chunkingEnabled": { "index": 34, "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": 35, "kind": "property", "displayName":
"Compression Type", "group": "producer", "label": "producer", "required":
false, "type": "object", "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", "descri [...]
- "initialSequenceId": { "index": 36, "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." },
- "lazyStartProducer": { "index": 37, "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": 38, "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": 39, "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 [...]
- "messageRouter": { "index": 40, "kind": "property", "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":
"configuration", "description": "Custom Message Router to use" },
- "messageRoutingMode": { "index": 41, "kind": "property", "displayName":
"Message Routing Mode", "group": "producer", "label": "producer", "required":
false, "type": "object", "javaType":
"org.apache.pulsar.client.api.MessageRoutingMode", "enum": [ "SinglePartition",
"RoundRobinPartition", "CustomPartition" ], "deprecated": false, "autowired":
false, "secret": false, "defaultValue": "RoundRobinPartition",
"configurationClass": "org.apache.camel.component.pulsar.PulsarConfiguration",
" [...]
- "producerName": { "index": 42, "kind": "property", "displayName":
"Producer Name", "group": "producer", "label": "producer", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"configuration", "description": "Name of the producer. If unset, lets Pulsar
select a unique identifier." },
- "sendTimeoutMs": { "index": 43, "kind": "property", "displayName": "Send
Timeout Ms", "group": "producer", "label": "producer", "required": false,
"type": "integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 30000, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"configuration", "description": "Send timeout in milliseconds" },
- "autoConfiguration": { "index": 44, "kind": "property", "displayName":
"Auto Configuration", "group": "advanced", "label": "advanced", "required":
false, "type": "object", "javaType":
"org.apache.camel.component.pulsar.utils.AutoConfiguration", "deprecated":
false, "autowired": false, "secret": false, "description": "The pulsar auto
configuration" },
- "autowiredEnabled": { "index": 45, "kind": "property", "displayName":
"Autowired Enabled", "group": "advanced", "label": "advanced", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true, "description":
"Whether autowiring is enabled. This is used for automatic autowiring options
(the option must be marked as autowired) by looking up in the registry to find
if there is a single instance of matching [...]
- "pulsarClient": { "index": 46, "kind": "property", "displayName": "Pulsar
Client", "group": "advanced", "label": "advanced", "required": false, "type":
"object", "javaType": "org.apache.pulsar.client.api.PulsarClient",
"deprecated": false, "autowired": true, "secret": false, "description": "The
pulsar client" }
+ "hashingScheme": { "index": 36, "kind": "property", "displayName":
"Hashing Scheme", "group": "producer", "label": "producer", "required": false,
"type": "string", "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 t [...]
+ "initialSequenceId": { "index": 37, "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." },
+ "lazyStartProducer": { "index": 38, "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": 39, "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": 40, "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 [...]
+ "messageRouter": { "index": 41, "kind": "property", "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":
"configuration", "description": "Custom Message Router to use" },
+ "messageRoutingMode": { "index": 42, "kind": "property", "displayName":
"Message Routing Mode", "group": "producer", "label": "producer", "required":
false, "type": "object", "javaType":
"org.apache.pulsar.client.api.MessageRoutingMode", "enum": [ "SinglePartition",
"RoundRobinPartition", "CustomPartition" ], "deprecated": false, "autowired":
false, "secret": false, "defaultValue": "RoundRobinPartition",
"configurationClass": "org.apache.camel.component.pulsar.PulsarConfiguration",
" [...]
+ "producerName": { "index": 43, "kind": "property", "displayName":
"Producer Name", "group": "producer", "label": "producer", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"configuration", "description": "Name of the producer. If unset, lets Pulsar
select a unique identifier." },
+ "sendTimeoutMs": { "index": 44, "kind": "property", "displayName": "Send
Timeout Ms", "group": "producer", "label": "producer", "required": false,
"type": "integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 30000, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"configuration", "description": "Send timeout in milliseconds" },
+ "autoConfiguration": { "index": 45, "kind": "property", "displayName":
"Auto Configuration", "group": "advanced", "label": "advanced", "required":
false, "type": "object", "javaType":
"org.apache.camel.component.pulsar.utils.AutoConfiguration", "deprecated":
false, "autowired": false, "secret": false, "description": "The pulsar auto
configuration" },
+ "autowiredEnabled": { "index": 46, "kind": "property", "displayName":
"Autowired Enabled", "group": "advanced", "label": "advanced", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true, "description":
"Whether autowiring is enabled. This is used for automatic autowiring options
(the option must be marked as autowired) by looking up in the registry to find
if there is a single instance of matching [...]
+ "pulsarClient": { "index": 47, "kind": "property", "displayName": "Pulsar
Client", "group": "advanced", "label": "advanced", "required": false, "type":
"object", "javaType": "org.apache.pulsar.client.api.PulsarClient",
"deprecated": false, "autowired": true, "secret": false, "description": "The
pulsar client" }
},
"headers": {
"properties": { "index": 0, "kind": "header", "displayName": "", "group":
"consumer", "label": "consumer", "required": false, "javaType": "Map<String,
String>", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The properties attached to the message.",
"constantName":
"org.apache.camel.component.pulsar.utils.message.PulsarMessageHeaders#PROPERTIES"
},
@@ -128,13 +129,14 @@
"blockIfQueueFull": { "index": 37, "kind": "parameter", "displayName":
"Block If Queue Full", "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": "Whether to block
the producing thread if pending messages queue is ful [...]
"chunkingEnabled": { "index": 38, "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": 39, "kind": "parameter", "displayName":
"Compression Type", "group": "producer", "label": "producer", "required":
false, "type": "object", "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", [...]
- "initialSequenceId": { "index": 40, "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." },
- "maxPendingMessages": { "index": 41, "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": 42, "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": 43, "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" },
- "messageRoutingMode": { "index": 44, "kind": "parameter", "displayName":
"Message Routing Mode", "group": "producer", "label": "producer", "required":
false, "type": "object", "javaType":
"org.apache.pulsar.client.api.MessageRoutingMode", "enum": [ "SinglePartition",
"RoundRobinPartition", "CustomPartition" ], "deprecated": false, "autowired":
false, "secret": false, "defaultValue": "RoundRobinPartition",
"configurationClass": "org.apache.camel.component.pulsar.PulsarConfiguration",
[...]
- "producerName": { "index": 45, "kind": "parameter", "displayName":
"Producer Name", "group": "producer", "label": "producer", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"pulsarConfiguration", "description": "Name of the producer. If unset, lets
Pulsar select a unique identifier." },
- "sendTimeoutMs": { "index": 46, "kind": "parameter", "displayName": "Send
Timeout Ms", "group": "producer", "label": "producer", "required": false,
"type": "integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 30000, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"pulsarConfiguration", "description": "Send timeout in milliseconds" },
- "lazyStartProducer": { "index": 47, "kind": "parameter", "displayName":
"Lazy Start Producer", "group": "producer (advanced)", "label":
"producer,advanced", "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 produ [...]
+ "hashingScheme": { "index": 40, "kind": "parameter", "displayName":
"Hashing Scheme", "group": "producer", "label": "producer", "required": false,
"type": "string", "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 fun [...]
+ "initialSequenceId": { "index": 41, "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." },
+ "maxPendingMessages": { "index": 42, "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": 43, "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": 44, "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" },
+ "messageRoutingMode": { "index": 45, "kind": "parameter", "displayName":
"Message Routing Mode", "group": "producer", "label": "producer", "required":
false, "type": "object", "javaType":
"org.apache.pulsar.client.api.MessageRoutingMode", "enum": [ "SinglePartition",
"RoundRobinPartition", "CustomPartition" ], "deprecated": false, "autowired":
false, "secret": false, "defaultValue": "RoundRobinPartition",
"configurationClass": "org.apache.camel.component.pulsar.PulsarConfiguration",
[...]
+ "producerName": { "index": 46, "kind": "parameter", "displayName":
"Producer Name", "group": "producer", "label": "producer", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"pulsarConfiguration", "description": "Name of the producer. If unset, lets
Pulsar select a unique identifier." },
+ "sendTimeoutMs": { "index": 47, "kind": "parameter", "displayName": "Send
Timeout Ms", "group": "producer", "label": "producer", "required": false,
"type": "integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 30000, "configurationClass":
"org.apache.camel.component.pulsar.PulsarConfiguration", "configurationField":
"pulsarConfiguration", "description": "Send timeout in milliseconds" },
+ "lazyStartProducer": { "index": 48, "kind": "parameter", "displayName":
"Lazy Start Producer", "group": "producer (advanced)", "label":
"producer,advanced", "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 produ [...]
}
}
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 e77fdd6c797..65bf6de0ba1 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
@@ -136,6 +136,10 @@ public class PulsarConfiguration implements Cloneable {
private MessageRoutingMode messageRoutingMode =
MessageRoutingMode.RoundRobinPartition;
@UriParam(label = "producer", description = "Custom Message Router to use")
private MessageRouter messageRouter;
+ @UriParam(label = "producer",
+ description = "Hashing function to use when choosing the
partition to use for a particular message",
+ enums = "JavaStringHash,Murmur3_32Hash", defaultValue =
"JavaStringHash")
+ private String hashingScheme = "JavaStringHash";
/**
* Returns a copy of this configuration
@@ -599,4 +603,15 @@ public class PulsarConfiguration implements Cloneable {
public void setAuthenticationParams(String authenticationParams) {
this.authenticationParams = authenticationParams;
}
+
+ public String getHashingScheme() {
+ return hashingScheme;
+ }
+
+ /**
+ * Hashing function to use when choosing the partition to use for a
particular message
+ */
+ public void setHashingScheme(String hashingScheme) {
+ this.hashingScheme = hashingScheme;
+ }
}
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 5637837ba7c..becd99a162f 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
@@ -31,10 +31,7 @@ import
org.apache.camel.component.pulsar.utils.message.PulsarMessageUtils;
import org.apache.camel.support.DefaultAsyncProducer;
import org.apache.camel.util.CastUtils;
import org.apache.camel.util.ObjectHelper;
-import org.apache.pulsar.client.api.Producer;
-import org.apache.pulsar.client.api.ProducerBuilder;
-import org.apache.pulsar.client.api.PulsarClientException;
-import org.apache.pulsar.client.api.TypedMessageBuilder;
+import org.apache.pulsar.client.api.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -138,6 +135,7 @@ public class PulsarProducer extends DefaultAsyncProducer {
.enableBatching(configuration.isBatchingEnabled()).batcherBuilder(configuration.getBatcherBuilder())
.initialSequenceId(configuration.getInitialSequenceId())
.compressionType(configuration.getCompressionType())
+
.hashingScheme(HashingScheme.valueOf(configuration.getHashingScheme()))
.enableChunking(configuration.isChunkingEnabled());
if (ObjectHelper.isNotEmpty(configuration.getMessageRouter()))
{
producerBuilder.messageRouter(configuration.getMessageRouter());
