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 f00daddb65a5 CAMEL-26012: camel-pulsar - the producer no longer sends
a body with Java serialization (#29753)
f00daddb65a5 is described below
commit f00daddb65a59c13a8b90cd8fdf56cbeb6805050
Author: Claus Ibsen <[email protected]>
AuthorDate: Fri Nov 13 08:05:05 2026 +0100
CAMEL-26012: camel-pulsar - the producer no longer sends a body with Java
serialization (#29753)
The pulsar producer fell back to Java serialization for a body that does not
convert to byte[] (a Map, List, number or POJO, and no body as a serialized
null). Now such a body, and no body, fail with an InvalidPayloadException
that
says to marshal it first. Text, bytes, streams and files are sent as before.
The new deprecated option allowJavaSerializedObject (insecure:serialization)
turns the old fallback back on. The ITs that sent an Integer now send a
String.
New "Message body" section on the component page and a 4.24 upgrade-guide
entry.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/components/pulsar.json | 12 +-
.../camel/catalog/docs/pulsar-component.adoc | 10 +
.../pulsar/PulsarComponentConfigurer.java | 6 +
.../component/pulsar/PulsarEndpointConfigurer.java | 6 +
.../component/pulsar/PulsarEndpointUriFactory.java | 3 +-
.../org/apache/camel/component/pulsar/pulsar.json | 12 +-
.../src/main/docs/pulsar-component.adoc | 10 +
.../component/pulsar/PulsarConfiguration.java | 20 ++
.../camel/component/pulsar/PulsarEndpoint.java | 6 +-
.../camel/component/pulsar/PulsarProducer.java | 35 ++--
.../component/pulsar/PulsarProducerBodyTest.java | 203 +++++++++++++++++++++
.../pulsar/integration/PulsarProducerInIT.java | 2 +-
.../integration/PulsarProducerURIClientIT.java | 2 +-
.../java/org/apache/camel/util/SecurityUtils.java | 1 +
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 11 ++
.../dsl/PulsarComponentBuilderFactory.java | 25 +++
.../endpoint/dsl/PulsarEndpointBuilderFactory.java | 44 +++++
17 files changed, 373 insertions(+), 35 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 1d239db34bbb..f2d26b095a8f 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
@@ -71,9 +71,10 @@
"messageRoutingMode": { "index": 44, "kind": "property", "displayName":
"Message Routing Mode", "group": "producer", "label": "producer", "required":
false, "type": "enum", "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",
"co [...]
"producerName": { "index": 45, "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": 46, "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": 47, "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": 48, "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": 49, "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" }
+ "allowJavaSerializedObject": { "index": 47, "kind": "property",
"displayName": "Allow Java Serialized Object", "group": "producer (advanced)",
"label": "producer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": true, "deprecationNote": "Java serialization of the
body is kept only for compatibility, and will be removed: marshal the body
before sending it", "autowired": false, "secret": false, "security":
"insecure:serialization", "defaultValue": fa [...]
+ "autoConfiguration": { "index": 48, "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": 49, "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": 50, "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, "autowired": false, "secret": false,
"description": "The properties attached to the message.", "constantName":
"org.apache.camel.component.pulsar.utils.message.PulsarMessageHeaders#PROPERTIES"
},
@@ -96,7 +97,7 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "bytes", "null" ], "javaType": "byte[]",
"description": "The payload of the Pulsar message as a byte array, null for a
message with a null value (a tombstone), with the key and the properties in
headers: convert or unmarshal it" },
- "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"description": "What converts to a byte array: text, bytes, a stream or a file,
sent as the payload. Any other body is sent with Java serialization, a Map, a
List, a number and no body too, and an object that is not Serializable fails:
marshal it first" },
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"dependsOn": [ "allowJavaSerializedObject" ], "description": "What converts to
a byte array: text, bytes, a stream or a file, sent as the payload. Any other
body, such as a Map, a List, a number or a POJO, fails, and so does no body:
marshal it first" },
"producerOut": { "carriers": [ "unchanged" ], "description": "Unchanged,
also when the exchange expects a reply, with the id of the published message in
a header" }
},
"properties": {
@@ -150,7 +151,8 @@
"messageRoutingMode": { "index": 47, "kind": "parameter", "displayName":
"Message Routing Mode", "group": "producer", "label": "producer", "required":
false, "type": "enum", "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",
"c [...]
"producerName": { "index": 48, "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": 49, "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": 50, "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 [...]
+ "allowJavaSerializedObject": { "index": 50, "kind": "parameter",
"displayName": "Allow Java Serialized Object", "group": "producer (advanced)",
"label": "producer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": true, "deprecationNote": "Java serialization of the
body is kept only for compatibility, and will be removed: marshal the body
before sending it", "autowired": false, "secret": false, "security":
"insecure:serialization", "defaultValue": f [...]
+ "lazyStartProducer": { "index": 51, "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 [...]
},
"dstWorlds": {
"PulsarWorld": { "javaType":
"org.apache.camel.component.pulsar.dst.PulsarWorld", "fake": "FakePulsar: a
broker and client for one topic and one durable subscription", "covers": [
"subscription types Exclusive, Shared, Failover and Key_Shared with one to
three consumers", "the message listener (messageListener=true, one consumer
thread)", "the ack timeout and the redelivery of a failed exchange after a
negative acknowledgement", "route start, restart, suspend and resume", "an
asynchr [...]
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 dd055e1772b3..0711ed91d912 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
@@ -73,6 +73,16 @@ The following routes send a message to the `orders` topic
every five seconds, an
message: "Received ${body}"
----
+=== Message body
+
+The producer sends the body as a byte array: text, bytes, a stream or a file
are sent as their content. A body that
+does not convert to a byte array, such as a `Map`, a `List`, a number or a
POJO, fails, and so does no body: marshal it
+first, for example to JSON with `marshal().json()`. The consumer gives the
payload as a byte array, so convert or
+unmarshal it in the route.
+
+Before Camel 4.24 such a body was sent with Java serialization, which only a
Java client that deserializes it can
+read. The deprecated `allowJavaSerializedObject` option turns this back on for
compatibility, and will be removed.
+
=== Producer sequence ids and deduplication
With a `producerName` and deduplication enabled on the broker, the producer
continues the sequence ids after the
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 e490cf635e72..dbf5e5185798 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
@@ -37,6 +37,8 @@ public class PulsarComponentConfigurer extends
PropertyConfigurerSupport impleme
case "ackTimeoutMillis":
getOrCreateConfiguration(target).setAckTimeoutMillis(property(camelContext,
long.class, value)); return true;
case "acktimeoutredeliverybackoff":
case "ackTimeoutRedeliveryBackoff":
getOrCreateConfiguration(target).setAckTimeoutRedeliveryBackoff(property(camelContext,
org.apache.pulsar.client.api.RedeliveryBackoff.class, value)); return true;
+ case "allowjavaserializedobject":
+ case "allowJavaSerializedObject":
getOrCreateConfiguration(target).setAllowJavaSerializedObject(property(camelContext,
boolean.class, value)); return true;
case "allowmanualacknowledgement":
case "allowManualAcknowledgement":
getOrCreateConfiguration(target).setAllowManualAcknowledgement(property(camelContext,
boolean.class, value)); return true;
case "authenticationclass":
@@ -148,6 +150,8 @@ public class PulsarComponentConfigurer extends
PropertyConfigurerSupport impleme
case "ackTimeoutMillis": return long.class;
case "acktimeoutredeliverybackoff":
case "ackTimeoutRedeliveryBackoff": return
org.apache.pulsar.client.api.RedeliveryBackoff.class;
+ case "allowjavaserializedobject":
+ case "allowJavaSerializedObject": return boolean.class;
case "allowmanualacknowledgement":
case "allowManualAcknowledgement": return boolean.class;
case "authenticationclass":
@@ -255,6 +259,8 @@ public class PulsarComponentConfigurer extends
PropertyConfigurerSupport impleme
case "ackTimeoutMillis": return
getOrCreateConfiguration(target).getAckTimeoutMillis();
case "acktimeoutredeliverybackoff":
case "ackTimeoutRedeliveryBackoff": return
getOrCreateConfiguration(target).getAckTimeoutRedeliveryBackoff();
+ case "allowjavaserializedobject":
+ case "allowJavaSerializedObject": return
getOrCreateConfiguration(target).isAllowJavaSerializedObject();
case "allowmanualacknowledgement":
case "allowManualAcknowledgement": return
getOrCreateConfiguration(target).isAllowManualAcknowledgement();
case "authenticationclass":
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 9a390aa59046..b4b1c1806fe8 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
@@ -30,6 +30,8 @@ public class PulsarEndpointConfigurer extends
PropertyConfigurerSupport implemen
case "ackTimeoutMillis":
target.getPulsarConfiguration().setAckTimeoutMillis(property(camelContext,
long.class, value)); return true;
case "acktimeoutredeliverybackoff":
case "ackTimeoutRedeliveryBackoff":
target.getPulsarConfiguration().setAckTimeoutRedeliveryBackoff(property(camelContext,
org.apache.pulsar.client.api.RedeliveryBackoff.class, value)); return true;
+ case "allowjavaserializedobject":
+ case "allowJavaSerializedObject":
target.getPulsarConfiguration().setAllowJavaSerializedObject(property(camelContext,
boolean.class, value)); return true;
case "allowmanualacknowledgement":
case "allowManualAcknowledgement":
target.getPulsarConfiguration().setAllowManualAcknowledgement(property(camelContext,
boolean.class, value)); return true;
case "authenticationclass":
@@ -131,6 +133,8 @@ public class PulsarEndpointConfigurer extends
PropertyConfigurerSupport implemen
case "ackTimeoutMillis": return long.class;
case "acktimeoutredeliverybackoff":
case "ackTimeoutRedeliveryBackoff": return
org.apache.pulsar.client.api.RedeliveryBackoff.class;
+ case "allowjavaserializedobject":
+ case "allowJavaSerializedObject": return boolean.class;
case "allowmanualacknowledgement":
case "allowManualAcknowledgement": return boolean.class;
case "authenticationclass":
@@ -233,6 +237,8 @@ public class PulsarEndpointConfigurer extends
PropertyConfigurerSupport implemen
case "ackTimeoutMillis": return
target.getPulsarConfiguration().getAckTimeoutMillis();
case "acktimeoutredeliverybackoff":
case "ackTimeoutRedeliveryBackoff": return
target.getPulsarConfiguration().getAckTimeoutRedeliveryBackoff();
+ case "allowjavaserializedobject":
+ case "allowJavaSerializedObject": return
target.getPulsarConfiguration().isAllowJavaSerializedObject();
case "allowmanualacknowledgement":
case "allowManualAcknowledgement": return
target.getPulsarConfiguration().isAllowManualAcknowledgement();
case "authenticationclass":
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 e8d947edb5a9..e41d972bf0a7 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
@@ -24,10 +24,11 @@ public class PulsarEndpointUriFactory extends
org.apache.camel.support.component
private static final Set<String> ENDPOINT_IDENTITY_PROPERTY_NAMES;
private static final Map<String, String> MULTI_VALUE_PREFIXES;
static {
- Set<String> props = new HashSet<>(51);
+ Set<String> props = new HashSet<>(52);
props.add("ackGroupTimeMillis");
props.add("ackTimeoutMillis");
props.add("ackTimeoutRedeliveryBackoff");
+ props.add("allowJavaSerializedObject");
props.add("allowManualAcknowledgement");
props.add("authenticationClass");
props.add("authenticationParams");
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 1d239db34bbb..f2d26b095a8f 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
@@ -71,9 +71,10 @@
"messageRoutingMode": { "index": 44, "kind": "property", "displayName":
"Message Routing Mode", "group": "producer", "label": "producer", "required":
false, "type": "enum", "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",
"co [...]
"producerName": { "index": 45, "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": 46, "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": 47, "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": 48, "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": 49, "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" }
+ "allowJavaSerializedObject": { "index": 47, "kind": "property",
"displayName": "Allow Java Serialized Object", "group": "producer (advanced)",
"label": "producer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": true, "deprecationNote": "Java serialization of the
body is kept only for compatibility, and will be removed: marshal the body
before sending it", "autowired": false, "secret": false, "security":
"insecure:serialization", "defaultValue": fa [...]
+ "autoConfiguration": { "index": 48, "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": 49, "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": 50, "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, "autowired": false, "secret": false,
"description": "The properties attached to the message.", "constantName":
"org.apache.camel.component.pulsar.utils.message.PulsarMessageHeaders#PROPERTIES"
},
@@ -96,7 +97,7 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "bytes", "null" ], "javaType": "byte[]",
"description": "The payload of the Pulsar message as a byte array, null for a
message with a null value (a tombstone), with the key and the properties in
headers: convert or unmarshal it" },
- "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"description": "What converts to a byte array: text, bytes, a stream or a file,
sent as the payload. Any other body is sent with Java serialization, a Map, a
List, a number and no body too, and an object that is not Serializable fails:
marshal it first" },
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"dependsOn": [ "allowJavaSerializedObject" ], "description": "What converts to
a byte array: text, bytes, a stream or a file, sent as the payload. Any other
body, such as a Map, a List, a number or a POJO, fails, and so does no body:
marshal it first" },
"producerOut": { "carriers": [ "unchanged" ], "description": "Unchanged,
also when the exchange expects a reply, with the id of the published message in
a header" }
},
"properties": {
@@ -150,7 +151,8 @@
"messageRoutingMode": { "index": 47, "kind": "parameter", "displayName":
"Message Routing Mode", "group": "producer", "label": "producer", "required":
false, "type": "enum", "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",
"c [...]
"producerName": { "index": 48, "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": 49, "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": 50, "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 [...]
+ "allowJavaSerializedObject": { "index": 50, "kind": "parameter",
"displayName": "Allow Java Serialized Object", "group": "producer (advanced)",
"label": "producer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": true, "deprecationNote": "Java serialization of the
body is kept only for compatibility, and will be removed: marshal the body
before sending it", "autowired": false, "secret": false, "security":
"insecure:serialization", "defaultValue": f [...]
+ "lazyStartProducer": { "index": 51, "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 [...]
},
"dstWorlds": {
"PulsarWorld": { "javaType":
"org.apache.camel.component.pulsar.dst.PulsarWorld", "fake": "FakePulsar: a
broker and client for one topic and one durable subscription", "covers": [
"subscription types Exclusive, Shared, Failover and Key_Shared with one to
three consumers", "the message listener (messageListener=true, one consumer
thread)", "the ack timeout and the redelivery of a failed exchange after a
negative acknowledgement", "route start, restart, suspend and resume", "an
asynchr [...]
diff --git a/components/camel-pulsar/src/main/docs/pulsar-component.adoc
b/components/camel-pulsar/src/main/docs/pulsar-component.adoc
index dd055e1772b3..0711ed91d912 100644
--- a/components/camel-pulsar/src/main/docs/pulsar-component.adoc
+++ b/components/camel-pulsar/src/main/docs/pulsar-component.adoc
@@ -73,6 +73,16 @@ The following routes send a message to the `orders` topic
every five seconds, an
message: "Received ${body}"
----
+=== Message body
+
+The producer sends the body as a byte array: text, bytes, a stream or a file
are sent as their content. A body that
+does not convert to a byte array, such as a `Map`, a `List`, a number or a
POJO, fails, and so does no body: marshal it
+first, for example to JSON with `marshal().json()`. The consumer gives the
payload as a byte array, so convert or
+unmarshal it in the route.
+
+Before Camel 4.24 such a body was sent with Java serialization, which only a
Java client that deserializes it can
+read. The deprecated `allowJavaSerializedObject` option turns this back on for
compatibility, and will be removed.
+
=== Producer sequence ids and deduplication
With a `producerName` and deduplication enabled on the broker, the producer
continues the sequence ids after the
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 12c2926be563..598d98ccffa8 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
@@ -22,6 +22,7 @@ import org.apache.camel.RuntimeCamelException;
import
org.apache.camel.component.pulsar.utils.consumers.SubscriptionInitialPosition;
import org.apache.camel.component.pulsar.utils.consumers.SubscriptionMode;
import org.apache.camel.component.pulsar.utils.consumers.SubscriptionType;
+import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriParams;
import org.apache.pulsar.client.api.BatcherBuilder;
@@ -158,6 +159,15 @@ public class PulsarConfiguration implements Cloneable {
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";
+ @UriParam(label = "producer,advanced", security = "insecure:serialization",
+ description = "Whether the producer sends a body that does not
convert to a byte array (such as a Map, a List,"
+ + " a number or a POJO) with Java serialization,
as Camel did before 4.24. Such a message can only"
+ + " be read by a Java client that deserializes it.
By default the producer fails such a body:"
+ + " marshal it first, for example to JSON.")
+ @Metadata(deprecationNote = "Java serialization of the body is kept only
for compatibility, and will be removed:"
+ + " marshal the body before sending it")
+ @Deprecated
+ private boolean allowJavaSerializedObject;
/**
* Returns a copy of this configuration
@@ -170,6 +180,16 @@ public class PulsarConfiguration implements Cloneable {
}
}
+ @Deprecated
+ public boolean isAllowJavaSerializedObject() {
+ return allowJavaSerializedObject;
+ }
+
+ @Deprecated
+ public void setAllowJavaSerializedObject(boolean
allowJavaSerializedObject) {
+ this.allowJavaSerializedObject = allowJavaSerializedObject;
+ }
+
public boolean isTopicsPattern() {
return topicsPattern;
}
diff --git
a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/PulsarEndpoint.java
b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/PulsarEndpoint.java
index fd5fdb1ee8c0..e11b289b8d22 100644
---
a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/PulsarEndpoint.java
+++
b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/PulsarEndpoint.java
@@ -46,9 +46,9 @@ import org.apache.pulsar.client.api.PulsarClient;
+ " convert or unmarshal it"),
producerIn = @BodyType(value = { Carrier.TEXT, Carrier.BYTES,
Carrier.STREAM, Carrier.FILE },
description = "What converts to a byte
array: text, bytes, a stream or a file, sent as the"
- + " payload. Any other body is
sent with Java serialization, a Map, a List, a"
- + " number and no body too,
and an object that is not Serializable fails:"
- + " marshal it first"),
+ + " payload. Any other body,
such as a Map, a List, a number or a POJO, fails,"
+ + " and so does no body:
marshal it first",
+ dependsOn = "allowJavaSerializedObject"),
producerOut = @BodyType(value = Carrier.UNCHANGED,
description = "Unchanged, also when the
exchange expects a reply, with the id of the"
+ " published message in a
header"))
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 7a40d2617d11..ac4e26e0f740 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
@@ -16,16 +16,14 @@
*/
package org.apache.camel.component.pulsar;
-import java.io.IOException;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.apache.camel.AsyncCallback;
import org.apache.camel.CamelExchangeException;
import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
import org.apache.camel.Message;
-import org.apache.camel.NoTypeConversionAvailableException;
-import org.apache.camel.TypeConversionException;
import org.apache.camel.component.pulsar.utils.message.PulsarMessageHeaders;
import org.apache.camel.component.pulsar.utils.message.PulsarMessageUtils;
import org.apache.camel.support.DefaultAsyncProducer;
@@ -103,24 +101,23 @@ public class PulsarProducer extends DefaultAsyncProducer {
}
/**
- * Serialize the given content using the appropriate converter if any,
otherwise it relies on
- * {@link PulsarMessageUtils#serialize(Object)}.
- *
- * @param exchange the exchange used as context for the serialization
process.
- * @param content the content to serialize.
- * @return the serialized counterpart of the given content
- * @throws IOException if an error occurs while serializing the content.
+ * The payload of the message: the body converted to a byte array. A body
that does not convert, such as a Map or a
+ * POJO, and no body, fail unless allowJavaSerializedObject is enabled,
which sends them with Java serialization.
*/
- private static byte[] serialize(Exchange exchange, Object content) throws
IOException {
- byte[] result;
- try {
- result = exchange.getContext().getTypeConverter()
- .mandatoryConvertTo(byte[].class, exchange, content);
- } catch (NoTypeConversionAvailableException | TypeConversionException
exception) {
- // fallback to try to serialize the data
- result = PulsarMessageUtils.serialize(content);
+ private byte[] serialize(Exchange exchange, Object content) throws
Exception {
+ if (content != null) {
+ byte[] result =
exchange.getContext().getTypeConverter().tryConvertTo(byte[].class, exchange,
content);
+ if (result != null) {
+ return result;
+ }
+ }
+ if
(pulsarEndpoint.getPulsarConfiguration().isAllowJavaSerializedObject()) {
+ return PulsarMessageUtils.serialize(content);
}
- return result;
+ String hint = content == null
+ ? "the pulsar producer does not send a message without a body"
+ : "the pulsar producer sends the body as a byte array: marshal
it first, for example with marshal().json()";
+ throw new InvalidPayloadException(exchange, byte[].class,
exchange.getIn(), hint);
}
private void createProducer() throws PulsarClientException {
diff --git
a/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/PulsarProducerBodyTest.java
b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/PulsarProducerBodyTest.java
new file mode 100644
index 000000000000..f00615ccfe0d
--- /dev/null
+++
b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/PulsarProducerBodyTest.java
@@ -0,0 +1,203 @@
+/*
+ * 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.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.ObjectInputStream;
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
+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.ProducerBuilder;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.TypedMessageBuilder;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Answers.RETURNS_SELF;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * The producer sends the body as a byte array. A body that does not convert
to one, and no body, fail unless the
+ * deprecated allowJavaSerializedObject option turns the Java serialization of
the body back on.
+ */
+public class PulsarProducerBodyTest extends CamelTestSupport {
+
+ private final List<byte[]> sent = new ArrayList<>();
+
+ @TempDir
+ Path tempDir;
+
+ @BeforeEach
+ void clearSent() {
+ sent.clear();
+ }
+
+ @Test
+ public void testTextIsSentAsItsBytes() {
+ Exchange out = template.send("direct:start", e ->
e.getIn().setBody("Hello World"));
+
+ assertNull(out.getException());
+ assertArrayEquals("Hello World".getBytes(StandardCharsets.UTF_8),
sent.get(0));
+ }
+
+ @Test
+ public void testBytesAreSentAsTheyAre() {
+ byte[] bytes = { 1, 2, 3 };
+ Exchange out = template.send("direct:start", e ->
e.getIn().setBody(bytes));
+
+ assertNull(out.getException());
+ assertArrayEquals(bytes, sent.get(0));
+ }
+
+ @Test
+ public void testStreamIsSentAsItsContent() {
+ Exchange out = template.send("direct:start",
+ e -> e.getIn().setBody(new ByteArrayInputStream("from a
stream".getBytes(StandardCharsets.UTF_8))));
+
+ assertNull(out.getException());
+ assertEquals("from a stream", new String(sent.get(0),
StandardCharsets.UTF_8));
+ }
+
+ @Test
+ public void testFileIsSentAsItsContent() throws Exception {
+ File file = Files.writeString(tempDir.resolve("order.json"),
"{\"id\":1}").toFile();
+ Exchange out = template.send("direct:start", e ->
e.getIn().setBody(file));
+
+ assertNull(out.getException());
+ assertEquals("{\"id\":1}", new String(sent.get(0),
StandardCharsets.UTF_8));
+ }
+
+ @Test
+ public void testMapFails() {
+ Map<String, Object> map = new LinkedHashMap<>();
+ map.put("id", 1);
+ Exchange out = template.send("direct:start", e ->
e.getIn().setBody(map));
+
+ InvalidPayloadException cause =
assertInstanceOf(InvalidPayloadException.class, out.getException());
+ assertTrue(cause.getMessage().contains("marshal it first"),
cause.getMessage());
+ assertTrue(sent.isEmpty(), "nothing should be sent");
+ }
+
+ @Test
+ public void testSerializablePojoFails() {
+ Exchange out = template.send("direct:start", e ->
e.getIn().setBody(new Order(42)));
+
+ InvalidPayloadException cause =
assertInstanceOf(InvalidPayloadException.class, out.getException());
+
assertTrue(cause.getMessage().contains(Order.class.getCanonicalName()),
cause.getMessage());
+ assertTrue(sent.isEmpty(), "nothing should be sent");
+ }
+
+ @Test
+ public void testNoBodyFails() {
+ Exchange out = template.send("direct:start", e ->
e.getIn().setBody(null));
+
+ InvalidPayloadException cause =
assertInstanceOf(InvalidPayloadException.class, out.getException());
+ assertTrue(cause.getMessage().contains("without a body"),
cause.getMessage());
+ assertTrue(sent.isEmpty(), "nothing should be sent");
+ }
+
+ @Test
+ public void testAllowJavaSerializedObjectSerializesAMap() throws Exception
{
+ Map<String, Object> map = new LinkedHashMap<>();
+ map.put("id", 1);
+ Exchange out = template.send("direct:legacy", e ->
e.getIn().setBody(map));
+
+ assertNull(out.getException());
+ byte[] payload = sent.get(0);
+ // the Java serialization stream magic
+ assertEquals((byte) 0xAC, payload[0]);
+ assertEquals((byte) 0xED, payload[1]);
+ try (ObjectInputStream in = new ObjectInputStream(new
ByteArrayInputStream(payload))) {
+ assertEquals(map, in.readObject());
+ }
+ }
+
+ @Test
+ public void testAllowJavaSerializedObjectStillSendsTextAsItsBytes() {
+ Exchange out = template.send("direct:legacy", e ->
e.getIn().setBody("Hello World"));
+
+ assertNull(out.getException());
+ assertArrayEquals("Hello World".getBytes(StandardCharsets.UTF_8),
sent.get(0));
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ final CamelContext context = super.createCamelContext();
+
+ final TypedMessageBuilder<byte[]> messageBuilder =
mock(TypedMessageBuilder.class);
+ when(messageBuilder.value(any())).thenAnswer(invocation -> {
+ sent.add(invocation.getArgument(0));
+ return messageBuilder;
+ });
+
when(messageBuilder.sendAsync()).thenReturn(CompletableFuture.completedFuture(mock(MessageId.class)));
+
+ final Producer<byte[]> pulsarProducer = mock(Producer.class);
+ when(pulsarProducer.newMessage()).thenReturn(messageBuilder);
+
+ final ProducerBuilder<byte[]> producerBuilder =
mock(ProducerBuilder.class, RETURNS_SELF);
+ when(producerBuilder.create()).thenReturn(pulsarProducer);
+
+ final PulsarClient pulsarClient = mock(PulsarClient.class);
+ when(pulsarClient.newProducer()).thenReturn(producerBuilder);
+
+ 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:start")
+
.to("pulsar:persistent://public/default/camel-body-test");
+ from("direct:legacy")
+
.to("pulsar:persistent://public/default/camel-body-legacy?allowJavaSerializedObject=true");
+ }
+ };
+ }
+
+ public record Order(int id) implements Serializable {
+ }
+}
diff --git
a/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/integration/PulsarProducerInIT.java
b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/integration/PulsarProducerInIT.java
index 678606bd5f20..f322ee471f34 100644
---
a/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/integration/PulsarProducerInIT.java
+++
b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/integration/PulsarProducerInIT.java
@@ -99,7 +99,7 @@ public class PulsarProducerInIT extends PulsarITSupport {
producerTemplate.sendBody("Hello ");
producerTemplate.sendBody("World ");
- producerTemplate.sendBody(10);
+ producerTemplate.sendBody("10");
MockEndpoint.assertIsSatisfied(10, TimeUnit.SECONDS, to);
}
diff --git
a/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/integration/PulsarProducerURIClientIT.java
b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/integration/PulsarProducerURIClientIT.java
index a5d62d549eea..a0cee5aaef49 100644
---
a/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/integration/PulsarProducerURIClientIT.java
+++
b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/integration/PulsarProducerURIClientIT.java
@@ -68,7 +68,7 @@ public class PulsarProducerURIClientIT extends
PulsarITSupport {
producerTemplate.sendBody("Hello ");
producerTemplate.sendBody("World ");
- producerTemplate.sendBody(10);
+ producerTemplate.sendBody("10");
MockEndpoint.assertIsSatisfied(10, TimeUnit.SECONDS, to);
}
diff --git
a/core/camel-util/src/main/java/org/apache/camel/util/SecurityUtils.java
b/core/camel-util/src/main/java/org/apache/camel/util/SecurityUtils.java
index 511d1600f11c..e896a8b1cadd 100644
--- a/core/camel-util/src/main/java/org/apache/camel/util/SecurityUtils.java
+++ b/core/camel-util/src/main/java/org/apache/camel/util/SecurityUtils.java
@@ -144,6 +144,7 @@ public final class SecurityUtils {
"component:http",
"component:https",
"component:jetty",
+ "component:pulsar",
"component:servlet",
"component:vertxhttp"));
owners.put("allowlocalwebhookurls", Set.of(
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 db656af4cbd1..df465eba457a 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
@@ -4735,6 +4735,17 @@ in retry mode: 16 redeliveries and its default retry
letter and dead letter topi
`retryLetterTopic` and `deadLetterTopic` only apply when `maxRedeliverCount`
is set. The consumer logs a WARN when it
starts with one of these topics but no `maxRedeliverCount` (and for
`retryLetterTopic` without `enableRetry`).
+=== camel-pulsar - the producer no longer sends a body with Java serialization
+
+The `pulsar` producer sends the body as a `byte[]`. A body that does not
convert to one, such as a `Map`, a `List`, a
+number or a POJO, was sent with Java serialization, which only a Java client
that deserializes it can read, and no body
+was sent as the bytes of a serialized `null`. Such a body, and no body, now
fail with an `InvalidPayloadException`.
+Text, bytes, a stream and a file are sent as their content, as before.
+
+Marshal the body before the producer, for example with `marshal().json()`. To
keep the old behaviour for now, set the
+new option `allowJavaSerializedObject=true` on the endpoint or the component.
The option is deprecated from the start
+and will be removed in a future release.
+
=== 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 fe389a937346..dd9f03ef4a6b 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
@@ -901,6 +901,30 @@ public interface PulsarComponentBuilderFactory {
return this;
}
+
+ /**
+ * Whether the producer sends a body that does not convert to a byte
+ * array (such as a Map, a List, a number or a POJO) with Java
+ * serialization, as Camel did before 4.24. Such a message can only be
+ * read by a Java client that deserializes it. By default the producer
+ * fails such a body: marshal it first, for example to JSON.
Deprecation
+ * note: Java serialization of the body is kept only for compatibility,
+ * and will be removed: marshal the body before sending it.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: producer (advanced)
+ *
+ * @param allowJavaSerializedObject the value to set
+ * @return the dsl builder
+ */
+ @Deprecated
+ default PulsarComponentBuilder allowJavaSerializedObject(boolean
allowJavaSerializedObject) {
+ doSetProperty("allowJavaSerializedObject",
allowJavaSerializedObject);
+ return this;
+ }
+
/**
* The pulsar auto configuration.
*
@@ -1023,6 +1047,7 @@ public interface PulsarComponentBuilderFactory {
case "messageRoutingMode":
getOrCreateConfiguration((PulsarComponent)
component).setMessageRoutingMode((org.apache.pulsar.client.api.MessageRoutingMode)
value); return true;
case "producerName": getOrCreateConfiguration((PulsarComponent)
component).setProducerName((java.lang.String) value); return true;
case "sendTimeoutMs": getOrCreateConfiguration((PulsarComponent)
component).setSendTimeoutMs((int) value); return true;
+ case "allowJavaSerializedObject":
getOrCreateConfiguration((PulsarComponent)
component).setAllowJavaSerializedObject((boolean) value); return true;
case "autoConfiguration": ((PulsarComponent)
component).setAutoConfiguration((org.apache.camel.component.pulsar.utils.AutoConfiguration)
value); return true;
case "autowiredEnabled": ((PulsarComponent)
component).setAutowiredEnabled((boolean) value); return true;
case "pulsarClient": ((PulsarComponent)
component).setPulsarClient((org.apache.pulsar.client.api.PulsarClient) value);
return true;
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 0d1106c79396..6f7460b4a776 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
@@ -1444,6 +1444,50 @@ public interface PulsarEndpointBuilderFactory {
return (PulsarEndpointProducerBuilder) this;
}
+ /**
+ * Whether the producer sends a body that does not convert to a byte
+ * array (such as a Map, a List, a number or a POJO) with Java
+ * serialization, as Camel did before 4.24. Such a message can only be
+ * read by a Java client that deserializes it. By default the producer
+ * fails such a body: marshal it first, for example to JSON.
Deprecation
+ * note: Java serialization of the body is kept only for compatibility,
+ * and will be removed: marshal the body before sending it.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: producer (advanced)
+ *
+ * @param allowJavaSerializedObject the value to set
+ * @return the dsl builder
+ */
+ @Deprecated
+ default AdvancedPulsarEndpointProducerBuilder
allowJavaSerializedObject(boolean allowJavaSerializedObject) {
+ doSetProperty("allowJavaSerializedObject",
allowJavaSerializedObject);
+ return this;
+ }
+ /**
+ * Whether the producer sends a body that does not convert to a byte
+ * array (such as a Map, a List, a number or a POJO) with Java
+ * serialization, as Camel did before 4.24. Such a message can only be
+ * read by a Java client that deserializes it. By default the producer
+ * fails such a body: marshal it first, for example to JSON.
Deprecation
+ * note: Java serialization of the body is kept only for compatibility,
+ * and will be removed: marshal the body before sending it.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: producer (advanced)
+ *
+ * @param allowJavaSerializedObject the value to set
+ * @return the dsl builder
+ */
+ @Deprecated
+ default AdvancedPulsarEndpointProducerBuilder
allowJavaSerializedObject(String allowJavaSerializedObject) {
+ doSetProperty("allowJavaSerializedObject",
allowJavaSerializedObject);
+ return this;
+ }
/**
* Whether the producer should be started lazy (on the first message).
* By starting lazy you can use this to allow CamelContext and routes
to
