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 7ee0d02c48d4 CAMEL-26194: camel-zookeeper - the producer waits for
every write, fails on a body it cannot store, and a DELETE never creates a
znode (#29020)
7ee0d02c48d4 is described below
commit 7ee0d02c48d4940bc3c2aec04fe4a42769671538
Author: Claus Ibsen <[email protected]>
AuthorDate: Tue Nov 3 11:43:10 2026 +0100
CAMEL-26194: camel-zookeeper - the producer waits for every write, fails on
a body it cannot store, and a DELETE never creates a znode (#29020)
…ils on a body it cannot store, and a DELETE never creates a znode
A body that did not convert to byte[] (a Map, a POJO, a number) overwrote
the
znode with null data; it now fails the exchange with an
InvalidPayloadException.
A null body still stores no data, and a DELETE does not read the body.
An InOnly write or delete was sent without waiting and its failure was only
logged; it now waits for the confirmation like InOut and fails the exchange.
A DELETE of a missing znode with create=true created it; it now fails with
NoNodeException. With listChildren=true a failed InOut write or delete was
replaced by the child listing; it now fails the exchange.
The repeat option and the docs say that without repeat=true the consumer
emits the data once and does not follow later changes.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../apache/camel/catalog/components/zookeeper.json | 8 +-
.../camel/catalog/docs/zookeeper-component.adoc | 15 +-
.../camel/component/zookeeper/zookeeper.json | 8 +-
.../src/main/docs/zookeeper-component.adoc | 15 +-
.../zookeeper/ZooKeeperConfiguration.java | 3 +-
.../component/zookeeper/ZooKeeperEndpoint.java | 10 +-
.../component/zookeeper/ZooKeeperProducer.java | 121 +++--------
.../camel/component/zookeeper/ZooKeeperUtils.java | 13 +-
.../component/zookeeper/ZooKeeperProducerTest.java | 230 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 16 ++
.../dsl/ZookeeperComponentBuilderFactory.java | 2 +
.../dsl/ZooKeeperEndpointBuilderFactory.java | 4 +
12 files changed, 330 insertions(+), 115 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/zookeeper.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/zookeeper.json
index 433a6fc5f69f..67c4c06e3c32 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/zookeeper.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/zookeeper.json
@@ -28,7 +28,7 @@
"timeout": { "index": 1, "kind": "property", "displayName": "Timeout",
"group": "common", "label": "", "required": false, "type": "integer",
"javaType": "int", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 5000, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "The time interval to
wait on connection before timing out." },
"backoff": { "index": 2, "kind": "property", "displayName": "Backoff",
"group": "consumer", "label": "consumer", "required": false, "type": "integer",
"javaType": "long", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 5000, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "The time interval to
backoff for after an error before retrying." },
"bridgeErrorHandler": { "index": 3, "kind": "property", "displayName":
"Bridge Error Handler", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false, "description":
"Allows for bridging the consumer to the Camel routing Error Handler, which
mean any exceptions (if possible) occurred while the Camel consumer is trying
to pickup incoming messages, or the like [...]
- "repeat": { "index": 4, "kind": "property", "displayName": "Repeat",
"group": "consumer", "label": "consumer", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "Should changes to the
znode be 'watched' and repeatedly processed." },
+ "repeat": { "index": 4, "kind": "property", "displayName": "Repeat",
"group": "consumer", "label": "consumer", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "Should changes to the
znode be 'watched' and repeatedly processed. Without it, the consumer emi [...]
"sendEmptyMessageOnDelete": { "index": 5, "kind": "property",
"displayName": "Send Empty Message On Delete", "group": "consumer", "label":
"consumer", "required": false, "type": "boolean", "javaType": "boolean",
"deprecated": false, "autowired": false, "secret": false, "defaultValue": true,
"configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "Upon the delete of a
znode, should an empty message be s [...]
"create": { "index": 6, "kind": "property", "displayName": "Create",
"group": "producer", "label": "producer", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "Should the endpoint
create the node if it does not currently exist." },
"createMode": { "index": 7, "kind": "property", "displayName": "Create
Mode", "group": "producer", "label": "producer", "required": false, "type":
"enum", "javaType": "java.lang.String", "enum": [ "PERSISTENT",
"PERSISTENT_SEQUENTIAL", "EPHEMERAL", "EPHEMERAL_SEQUENTIAL" ], "deprecated":
false, "autowired": false, "secret": false, "defaultValue": "EPHEMERAL",
"configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "d [...]
@@ -47,8 +47,8 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "bytes", "null" ], "dependsOn": [
"sendEmptyMessageOnDelete" ], "byOption": [ { "option": "listChildren",
"value": "true", "carriers": [ "list" ] } ], "description": "The data of the
znode as bytes, and no body when the znode is deleted; with listChildren, the
List of the names of its child znodes. The CamelZooKeeperNode and
CamelZookeeperStatistics headers say which znode and its version" },
- "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"description": "The data to store in the znode, converted to bytes. A Map, a
POJO or a number is not converted and stores no data; with the
CamelZookeeperOperation header set to DELETE the body is not read" },
- "producerOut": { "carriers": [ "unchanged", "bytes", "text", "value" ],
"byOption": [ { "option": "listChildren", "value": "true", "carriers": [
"list", "unchanged" ] } ], "description": "InOnly writes without waiting and
leaves the body unchanged. InOut waits and sets the bytes written, the path of
the znode as text when it was created, or true for a DELETE; with listChildren,
the List of the names of the child znodes of the path" }
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"description": "The data to store in the znode, converted to bytes; a null body
stores no data. A Map, a POJO or a number is not converted and fails the
exchange; with the CamelZookeeperOperation header set to DELETE the body is not
read" },
+ "producerOut": { "carriers": [ "unchanged", "bytes", "text", "value" ],
"byOption": [ { "option": "listChildren", "value": "true", "carriers": [
"list", "unchanged" ] } ], "description": "Both wait for the write or delete,
and fail the exchange when it failed. InOnly leaves the body unchanged. InOut
sets the bytes written, the path of the znode as text when it was created, or
true for a DELETE; with listChildren, the List of the names of the child znodes
of the path" }
},
"properties": {
"serverUrls": { "index": 0, "kind": "path", "displayName": "Server Urls",
"group": "common", "label": "", "required": true, "type": "string", "javaType":
"java.lang.String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "The zookeeper server
hosts (multiple servers can be separated by comma)" },
@@ -56,7 +56,7 @@
"listChildren": { "index": 2, "kind": "parameter", "displayName": "List
Children", "group": "common", "label": "", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "Whether the children of
the node should be listed" },
"timeout": { "index": 3, "kind": "parameter", "displayName": "Timeout",
"group": "common", "label": "", "required": false, "type": "integer",
"javaType": "int", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 5000, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "The time interval to
wait on connection before timing out." },
"backoff": { "index": 4, "kind": "parameter", "displayName": "Backoff",
"group": "consumer", "label": "consumer", "required": false, "type": "integer",
"javaType": "long", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 5000, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "The time interval to
backoff for after an error before retrying." },
- "repeat": { "index": 5, "kind": "parameter", "displayName": "Repeat",
"group": "consumer", "label": "consumer", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "Should changes to the
znode be 'watched' and repeatedly processed." },
+ "repeat": { "index": 5, "kind": "parameter", "displayName": "Repeat",
"group": "consumer", "label": "consumer", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "Should changes to the
znode be 'watched' and repeatedly processed. Without it, the consumer em [...]
"sendEmptyMessageOnDelete": { "index": 6, "kind": "parameter",
"displayName": "Send Empty Message On Delete", "group": "consumer", "label":
"consumer", "required": false, "type": "boolean", "javaType": "boolean",
"deprecated": false, "autowired": false, "secret": false, "defaultValue": true,
"configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "Upon the delete of a
znode, should an empty message be [...]
"bridgeErrorHandler": { "index": 7, "kind": "parameter", "displayName":
"Bridge Error Handler", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Allows for bridging the consumer to the
Camel routing Error Handler, which mean any exceptions (if possible) occurred
while the Camel consumer is trying to pickup incoming [...]
"exceptionHandler": { "index": 8, "kind": "parameter", "displayName":
"Exception Handler", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "object", "javaType":
"org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.",
"deprecated": false, "autowired": false, "secret": false, "description": "To
let the consumer use a custom ExceptionHandler. Notice if the option
bridgeErrorHandler is enabled then this option is not in use. By def [...]
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/zookeeper-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/zookeeper-component.adoc
index ad2212dc850d..ed64a415c6ad 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/zookeeper-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/zookeeper-component.adoc
@@ -94,6 +94,10 @@ YAML::
If the node does not exist yet, the consumer waits for it to be created before
reading it.
+The consumer reads the data of the node (or its list of children with
`listChildren=true`) once, and does not
+follow later changes. To receive each change as well, set `repeat=true`, for
example
+`zookeeper://localhost:39913/somepath/somenode?repeat=true`.
+
=== Reading from a _znode_
When data is read due to a `WatchedEvent` received from the ZooKeeper
@@ -150,7 +154,8 @@ not to `/somepath/somenode` but to the path from the header
`/somepath/someothernode`.
WARNING: the `testPayload` must be convertible
-to `byte[]` as the data stored in ZooKeeper is byte-based.
+to `byte[]` as the data stored in ZooKeeper is byte-based. A body that cannot
be converted, such as a `Map`, a POJO
+or a number, fails the exchange with an `InvalidPayloadException`. A null body
stores no data in the _znode_.
._Java-only: ProducerTemplate with dynamic znode header_
[source,java]
@@ -197,8 +202,14 @@ YAML::
----
====
+The producer waits for ZooKeeper to confirm the write, for an InOnly exchange
too, and a failed write fails the
+exchange: for example a _znode_ that does not exist without the `create`
option, or a `CamelZooKeeperVersion` header
+that is not the current version of the _znode_. An InOnly exchange keeps its
body; an InOut exchange gets the result
+of the write as its body.
+
It is also possible to *delete* a node using the
-header `CamelZookeeperOperation` by setting it to `DELETE`:
+header `CamelZookeeperOperation` by setting it to `DELETE`. A delete of a
_znode_ that does not exist fails the
+exchange with a `NoNodeException`; the `create` option does not apply to a
delete:
[tabs]
====
diff --git
a/components/camel-zookeeper/src/generated/resources/META-INF/org/apache/camel/component/zookeeper/zookeeper.json
b/components/camel-zookeeper/src/generated/resources/META-INF/org/apache/camel/component/zookeeper/zookeeper.json
index 433a6fc5f69f..67c4c06e3c32 100644
---
a/components/camel-zookeeper/src/generated/resources/META-INF/org/apache/camel/component/zookeeper/zookeeper.json
+++
b/components/camel-zookeeper/src/generated/resources/META-INF/org/apache/camel/component/zookeeper/zookeeper.json
@@ -28,7 +28,7 @@
"timeout": { "index": 1, "kind": "property", "displayName": "Timeout",
"group": "common", "label": "", "required": false, "type": "integer",
"javaType": "int", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 5000, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "The time interval to
wait on connection before timing out." },
"backoff": { "index": 2, "kind": "property", "displayName": "Backoff",
"group": "consumer", "label": "consumer", "required": false, "type": "integer",
"javaType": "long", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 5000, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "The time interval to
backoff for after an error before retrying." },
"bridgeErrorHandler": { "index": 3, "kind": "property", "displayName":
"Bridge Error Handler", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false, "description":
"Allows for bridging the consumer to the Camel routing Error Handler, which
mean any exceptions (if possible) occurred while the Camel consumer is trying
to pickup incoming messages, or the like [...]
- "repeat": { "index": 4, "kind": "property", "displayName": "Repeat",
"group": "consumer", "label": "consumer", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "Should changes to the
znode be 'watched' and repeatedly processed." },
+ "repeat": { "index": 4, "kind": "property", "displayName": "Repeat",
"group": "consumer", "label": "consumer", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "Should changes to the
znode be 'watched' and repeatedly processed. Without it, the consumer emi [...]
"sendEmptyMessageOnDelete": { "index": 5, "kind": "property",
"displayName": "Send Empty Message On Delete", "group": "consumer", "label":
"consumer", "required": false, "type": "boolean", "javaType": "boolean",
"deprecated": false, "autowired": false, "secret": false, "defaultValue": true,
"configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "Upon the delete of a
znode, should an empty message be s [...]
"create": { "index": 6, "kind": "property", "displayName": "Create",
"group": "producer", "label": "producer", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "Should the endpoint
create the node if it does not currently exist." },
"createMode": { "index": 7, "kind": "property", "displayName": "Create
Mode", "group": "producer", "label": "producer", "required": false, "type":
"enum", "javaType": "java.lang.String", "enum": [ "PERSISTENT",
"PERSISTENT_SEQUENTIAL", "EPHEMERAL", "EPHEMERAL_SEQUENTIAL" ], "deprecated":
false, "autowired": false, "secret": false, "defaultValue": "EPHEMERAL",
"configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "d [...]
@@ -47,8 +47,8 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "bytes", "null" ], "dependsOn": [
"sendEmptyMessageOnDelete" ], "byOption": [ { "option": "listChildren",
"value": "true", "carriers": [ "list" ] } ], "description": "The data of the
znode as bytes, and no body when the znode is deleted; with listChildren, the
List of the names of its child znodes. The CamelZooKeeperNode and
CamelZookeeperStatistics headers say which znode and its version" },
- "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"description": "The data to store in the znode, converted to bytes. A Map, a
POJO or a number is not converted and stores no data; with the
CamelZookeeperOperation header set to DELETE the body is not read" },
- "producerOut": { "carriers": [ "unchanged", "bytes", "text", "value" ],
"byOption": [ { "option": "listChildren", "value": "true", "carriers": [
"list", "unchanged" ] } ], "description": "InOnly writes without waiting and
leaves the body unchanged. InOut waits and sets the bytes written, the path of
the znode as text when it was created, or true for a DELETE; with listChildren,
the List of the names of the child znodes of the path" }
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"description": "The data to store in the znode, converted to bytes; a null body
stores no data. A Map, a POJO or a number is not converted and fails the
exchange; with the CamelZookeeperOperation header set to DELETE the body is not
read" },
+ "producerOut": { "carriers": [ "unchanged", "bytes", "text", "value" ],
"byOption": [ { "option": "listChildren", "value": "true", "carriers": [
"list", "unchanged" ] } ], "description": "Both wait for the write or delete,
and fail the exchange when it failed. InOnly leaves the body unchanged. InOut
sets the bytes written, the path of the znode as text when it was created, or
true for a DELETE; with listChildren, the List of the names of the child znodes
of the path" }
},
"properties": {
"serverUrls": { "index": 0, "kind": "path", "displayName": "Server Urls",
"group": "common", "label": "", "required": true, "type": "string", "javaType":
"java.lang.String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "The zookeeper server
hosts (multiple servers can be separated by comma)" },
@@ -56,7 +56,7 @@
"listChildren": { "index": 2, "kind": "parameter", "displayName": "List
Children", "group": "common", "label": "", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "Whether the children of
the node should be listed" },
"timeout": { "index": 3, "kind": "parameter", "displayName": "Timeout",
"group": "common", "label": "", "required": false, "type": "integer",
"javaType": "int", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 5000, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "The time interval to
wait on connection before timing out." },
"backoff": { "index": 4, "kind": "parameter", "displayName": "Backoff",
"group": "consumer", "label": "consumer", "required": false, "type": "integer",
"javaType": "long", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 5000, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "The time interval to
backoff for after an error before retrying." },
- "repeat": { "index": 5, "kind": "parameter", "displayName": "Repeat",
"group": "consumer", "label": "consumer", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "Should changes to the
znode be 'watched' and repeatedly processed." },
+ "repeat": { "index": 5, "kind": "parameter", "displayName": "Repeat",
"group": "consumer", "label": "consumer", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "Should changes to the
znode be 'watched' and repeatedly processed. Without it, the consumer em [...]
"sendEmptyMessageOnDelete": { "index": 6, "kind": "parameter",
"displayName": "Send Empty Message On Delete", "group": "consumer", "label":
"consumer", "required": false, "type": "boolean", "javaType": "boolean",
"deprecated": false, "autowired": false, "secret": false, "defaultValue": true,
"configurationClass":
"org.apache.camel.component.zookeeper.ZooKeeperConfiguration",
"configurationField": "configuration", "description": "Upon the delete of a
znode, should an empty message be [...]
"bridgeErrorHandler": { "index": 7, "kind": "parameter", "displayName":
"Bridge Error Handler", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Allows for bridging the consumer to the
Camel routing Error Handler, which mean any exceptions (if possible) occurred
while the Camel consumer is trying to pickup incoming [...]
"exceptionHandler": { "index": 8, "kind": "parameter", "displayName":
"Exception Handler", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "object", "javaType":
"org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.",
"deprecated": false, "autowired": false, "secret": false, "description": "To
let the consumer use a custom ExceptionHandler. Notice if the option
bridgeErrorHandler is enabled then this option is not in use. By def [...]
diff --git a/components/camel-zookeeper/src/main/docs/zookeeper-component.adoc
b/components/camel-zookeeper/src/main/docs/zookeeper-component.adoc
index ad2212dc850d..ed64a415c6ad 100644
--- a/components/camel-zookeeper/src/main/docs/zookeeper-component.adoc
+++ b/components/camel-zookeeper/src/main/docs/zookeeper-component.adoc
@@ -94,6 +94,10 @@ YAML::
If the node does not exist yet, the consumer waits for it to be created before
reading it.
+The consumer reads the data of the node (or its list of children with
`listChildren=true`) once, and does not
+follow later changes. To receive each change as well, set `repeat=true`, for
example
+`zookeeper://localhost:39913/somepath/somenode?repeat=true`.
+
=== Reading from a _znode_
When data is read due to a `WatchedEvent` received from the ZooKeeper
@@ -150,7 +154,8 @@ not to `/somepath/somenode` but to the path from the header
`/somepath/someothernode`.
WARNING: the `testPayload` must be convertible
-to `byte[]` as the data stored in ZooKeeper is byte-based.
+to `byte[]` as the data stored in ZooKeeper is byte-based. A body that cannot
be converted, such as a `Map`, a POJO
+or a number, fails the exchange with an `InvalidPayloadException`. A null body
stores no data in the _znode_.
._Java-only: ProducerTemplate with dynamic znode header_
[source,java]
@@ -197,8 +202,14 @@ YAML::
----
====
+The producer waits for ZooKeeper to confirm the write, for an InOnly exchange
too, and a failed write fails the
+exchange: for example a _znode_ that does not exist without the `create`
option, or a `CamelZooKeeperVersion` header
+that is not the current version of the _znode_. An InOnly exchange keeps its
body; an InOut exchange gets the result
+of the write as its body.
+
It is also possible to *delete* a node using the
-header `CamelZookeeperOperation` by setting it to `DELETE`:
+header `CamelZookeeperOperation` by setting it to `DELETE`. A delete of a
_znode_ that does not exist fails the
+exchange with a `NoNodeException`; the `create` option does not apply to a
delete:
[tabs]
====
diff --git
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConfiguration.java
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConfiguration.java
index 9922d564f785..9b098f600ad7 100644
---
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConfiguration.java
+++
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConfiguration.java
@@ -145,7 +145,8 @@ public class ZooKeeperConfiguration implements Cloneable {
}
/**
- * Should changes to the znode be 'watched' and repeatedly processed.
+ * Should changes to the znode be 'watched' and repeatedly processed.
Without it, the consumer emits the data (or
+ * the child listing) of the znode once and does not follow later changes.
*/
public void setRepeat(boolean repeat) {
this.repeat = repeat;
diff --git
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperEndpoint.java
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperEndpoint.java
index 97de9258c2eb..ad56684f752a 100644
---
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperEndpoint.java
+++
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperEndpoint.java
@@ -47,14 +47,14 @@ import org.apache.camel.support.DefaultEndpoint;
+ " CamelZooKeeperNode and
CamelZookeeperStatistics headers say which znode"
+ " and its version"),
producerIn = @BodyType(value = { Carrier.TEXT, Carrier.BYTES,
Carrier.STREAM, Carrier.FILE },
- description = "The data to store in the
znode, converted to bytes. A Map, a POJO or a"
- + " number is not converted
and stores no data; with the CamelZookeeperOperation"
- + " header set to DELETE the
body is not read"),
+ description = "The data to store in the
znode, converted to bytes; a null body stores"
+ + " no data. A Map, a POJO or
a number is not converted and fails the exchange;"
+ + " with the
CamelZookeeperOperation header set to DELETE the body is not read"),
producerOut = @BodyType(value = { Carrier.UNCHANGED, Carrier.BYTES,
Carrier.TEXT, Carrier.VALUE },
byOption = @BodyType.When(option =
"listChildren", value = "true",
carriers = {
Carrier.LIST, Carrier.UNCHANGED }),
- description = "InOnly writes without
waiting and leaves the body unchanged. InOut waits and"
- + " sets the bytes written,
the path of the znode as text when it was created,"
+ description = "Both wait for the write or
delete, and fail the exchange when it failed."
+ + " InOnly leaves the body
unchanged. InOut sets the bytes written, the path of the znode as text when it
was created,"
+ " or true for a DELETE;
with listChildren, the List of the names of the child"
+ " znodes of the path"))
public class ZooKeeperEndpoint extends DefaultEndpoint implements
EndpointServiceLocation {
diff --git
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperProducer.java
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperProducer.java
index e8c01cba654c..040871f0fde0 100644
---
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperProducer.java
+++
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperProducer.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.zookeeper;
import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
import org.apache.camel.Message;
import org.apache.camel.component.zookeeper.operations.CreateOperation;
import org.apache.camel.component.zookeeper.operations.DeleteOperation;
@@ -25,12 +26,9 @@ import
org.apache.camel.component.zookeeper.operations.OperationResult;
import org.apache.camel.component.zookeeper.operations.SetDataOperation;
import org.apache.camel.support.DefaultProducer;
import org.apache.camel.support.ExchangeHelper;
-import org.apache.zookeeper.AsyncCallback.StatCallback;
-import org.apache.zookeeper.AsyncCallback.VoidCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -70,37 +68,30 @@ public class ZooKeeperProducer extends DefaultProducer {
// handle it got here, as other threads may send concurrently
ZooKeeper zk = this.zkm.getConnection();
connection = zk;
- ProductionContext context = new ProductionContext(zk, exchange);
String operation =
exchange.getIn().getHeader(ZooKeeperMessage.ZOOKEEPER_OPERATION, String.class);
boolean isDelete = ZK_OPERATION_DELETE.equals(operation);
+ ProductionContext context = new ProductionContext(zk, exchange,
isDelete);
- if (ExchangeHelper.isOutCapable(exchange)) {
- if (isDelete) {
- LOG.debug("Deleting znode '{}', waiting for confirmation",
context.node);
-
- OperationResult result = synchronouslyDelete(context);
- if (configuration.isListChildren()) {
- result = listChildren(context);
- }
- updateExchangeWithResult(context, result);
- } else {
- LOG.debug("Storing data to znode '{}', waiting for
confirmation", context.node);
-
- OperationResult result = synchronouslySetData(context);
- if (configuration.isListChildren()) {
- result = listChildren(context);
- }
- updateExchangeWithResult(context, result);
- }
+ // InOnly waits for the confirmation as well, so that a failed write
or delete fails the exchange
+ OperationResult result;
+ if (isDelete) {
+ LOG.debug("Deleting znode '{}', waiting for confirmation",
context.node);
+ result = synchronouslyDelete(context);
} else {
- if (isDelete) {
- asynchronouslyDeleteNode(zk, context);
- } else {
- asynchronouslySetDataOnNode(zk, context);
- }
+ LOG.debug("Storing data to znode '{}', waiting for confirmation",
context.node);
+ result = synchronouslySetData(context);
}
+ if (ExchangeHelper.isOutCapable(exchange)) {
+ // the listing of the children must not hide a failed write or
delete
+ if (result.isOk() && configuration.isListChildren()) {
+ result = listChildren(context);
+ }
+ updateExchangeWithResult(context, result);
+ } else if (!result.isOk()) {
+ exchange.setException(result.getException());
+ }
}
@Override
@@ -120,19 +111,6 @@ public class ZooKeeperProducer extends DefaultProducer {
zkm.shutdown();
}
- private void asynchronouslyDeleteNode(ZooKeeper connection,
ProductionContext context) {
- LOG.debug("Deleting node '{}', not waiting for confirmation",
context.node);
-
- connection.delete(context.node, context.version, new
AsyncDeleteCallback(), context);
-
- }
-
- private void asynchronouslySetDataOnNode(ZooKeeper connection,
ProductionContext context) {
- LOG.debug("Storing data to node '{}', not waiting for confirmation",
context.node);
-
- connection.setData(context.node, context.payload, context.version, new
AsyncSetDataCallback(), context);
- }
-
private void updateExchangeWithResult(ProductionContext context,
OperationResult result) {
ZooKeeperMessage out = new ZooKeeperMessage(
getEndpoint().getCamelContext(), context.node,
result.getStatistics(), context.in.getHeaders());
@@ -158,45 +136,14 @@ public class ZooKeeperProducer extends DefaultProducer {
int version;
String node;
- ProductionContext(ZooKeeper connection, Exchange exchange) {
+ ProductionContext(ZooKeeper connection, Exchange exchange, boolean
isDelete) throws InvalidPayloadException {
this.connection = connection;
this.exchange = exchange;
this.in = exchange.getIn();
this.node = getNodeFromMessage(in, configuration.getPath());
this.version = getVersionFromMessage(in);
- this.payload = getPayloadFromExchange(exchange);
- }
- }
-
- private class AsyncSetDataCallback implements StatCallback {
-
- @Override
- public void processResult(int rc, String node, Object ctx, Stat
statistics) {
- if (Code.NONODE.equals(Code.get(rc))) {
- if (configuration.isCreate()) {
- LOG.warn("Node '{}' did not exist, creating it...", node);
- ProductionContext context = (ProductionContext) ctx;
- OperationResult<String> result = null;
- try {
- result = createNode(context);
- } catch (Exception e) {
- LOG.error("Error trying to create node '{}'", node, e);
- }
-
- if (result == null || !result.isOk()) {
- LOG.error("Error creating node '{}'", node,
result.getException());
- }
- }
- } else {
- logStoreComplete(node, statistics);
- }
- }
- }
-
- private class AsyncDeleteCallback implements VoidCallback {
- @Override
- public void processResult(int rc, String path, Object ctx) {
- LOG.debug("Removed data node '{}'", path);
+ // a DELETE does not read the body
+ this.payload = isDelete ? null : getPayloadFromExchange(exchange);
}
}
@@ -236,26 +183,12 @@ public class ZooKeeperProducer extends DefaultProducer {
return result;
}
+ /**
+ * Deletes the node; a node that does not exist is never created (the
create option only applies to writes).
+ */
private OperationResult synchronouslyDelete(ProductionContext ctx) throws
Exception {
- DeleteOperation setData = new DeleteOperation(ctx.connection,
ctx.node);
- setData.setVersion(ctx.version);
-
- OperationResult result = setData.get();
-
- if (!result.isOk() && configuration.isCreate() &&
result.failedDueTo(Code.NONODE)) {
- LOG.warn("Node '{}' did not exist, creating it.", ctx.node);
- result = createNode(ctx);
- }
- return result;
- }
-
- private void logStoreComplete(String path, Stat statistics) {
- if (LOG.isDebugEnabled()) {
- if (LOG.isTraceEnabled()) {
- LOG.trace("Stored data to node '{}', and receive statistics
{}", path, statistics);
- } else {
- LOG.debug("Stored data to node '{}'", path);
- }
- }
+ DeleteOperation delete = new DeleteOperation(ctx.connection, ctx.node);
+ delete.setVersion(ctx.version);
+ return delete.get();
}
}
diff --git
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperUtils.java
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperUtils.java
index a51b7795a48b..ce0639c59f4d 100644
---
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperUtils.java
+++
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperUtils.java
@@ -19,10 +19,10 @@ package org.apache.camel.component.zookeeper;
import java.util.List;
import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
import org.apache.camel.Message;
import org.apache.camel.component.zookeeper.operations.WatchedEventProvider;
import org.apache.camel.component.zookeeper.operations.ZooKeeperOperation;
-import org.apache.camel.support.ExchangeHelper;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.ZooDefs.Ids;
@@ -86,8 +86,15 @@ public final class ZooKeeperUtils {
return getZookeeperProperty(message,
ZooKeeperMessage.ZOOKEEPER_NODE_VERSION, -1, Integer.class);
}
- public static byte[] getPayloadFromExchange(Exchange exchange) {
- return ExchangeHelper.convertToType(exchange, byte[].class,
exchange.getIn().getBody());
+ /**
+ * The body of the message as the data to store in a znode. A null body is
stored as no data; a body that cannot be
+ * converted to bytes fails rather than storing no data.
+ *
+ * @throws InvalidPayloadException if the body cannot be converted to bytes
+ */
+ public static byte[] getPayloadFromExchange(Exchange exchange) throws
InvalidPayloadException {
+ Message in = exchange.getIn();
+ return in.getBody() == null ? null : in.getMandatoryBody(byte[].class);
}
@SuppressWarnings("unchecked")
diff --git
a/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ZooKeeperProducerTest.java
b/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ZooKeeperProducerTest.java
new file mode 100644
index 000000000000..5e90c3a7d603
--- /dev/null
+++
b/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ZooKeeperProducerTest.java
@@ -0,0 +1,230 @@
+/*
+ * 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.zookeeper;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.InvalidPayloadException;
+import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Watcher.Event.KeeperState;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.server.NIOServerCnxnFactory;
+import org.apache.zookeeper.server.ZooKeeperServer;
+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;
+
+/**
+ * The ZooKeeper producer against a ZooKeeper server that runs in the test
JVM: the body that is stored, and the
+ * failures of InOnly and InOut writes and deletes.
+ */
+class ZooKeeperProducerTest extends CamelTestSupport {
+
+ private static final String NODE = "/camel-producer";
+ private static final String PARENT = "/camel-producer-parent";
+ private static final String MISSING = "/camel-producer-missing";
+
+ @TempDir
+ Path dataDir;
+
+ private final int port = AvailablePortFinder.getNextAvailable();
+
+ private ZooKeeperServer server;
+ private NIOServerCnxnFactory factory;
+ private ZooKeeper client;
+
+ @Override
+ protected void doPreSetup() throws Exception {
+ File dir = dataDir.toFile();
+ server = new ZooKeeperServer(dir, dir, 2000);
+ factory = new NIOServerCnxnFactory();
+ factory.configure(new InetSocketAddress("127.0.0.1", port), 60);
+ factory.startup(server);
+
+ CountDownLatch connected = new CountDownLatch(1);
+ client = new ZooKeeper("127.0.0.1:" + port, 20000, e -> {
+ if (e.getState() == KeeperState.SyncConnected) {
+ connected.countDown();
+ }
+ });
+ assertTrue(connected.await(20, TimeUnit.SECONDS), "the test client did
not connect");
+ client.create(NODE, bytes("one"), Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);
+ client.create(PARENT, bytes("parent"), Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);
+ client.create(PARENT + "/child", bytes("child"), Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);
+ }
+
+ @Override
+ protected void doPostTearDown() throws Exception {
+ if (client != null) {
+ client.close();
+ }
+ if (factory != null) {
+ factory.shutdown();
+ }
+ if (server != null) {
+ server.shutdown();
+ }
+ }
+
+ @Test
+ void mapBodyFailsAndKeepsTheData() throws Exception {
+ Exchange exchange = send(ExchangePattern.InOnly, uri(NODE),
Map.of("key", "value"), Map.of());
+
+ assertInstanceOf(InvalidPayloadException.class,
exchange.getException());
+ assertEquals("one", data(NODE));
+ }
+
+ @Test
+ void pojoBodyFailsAndKeepsTheData() throws Exception {
+ Exchange exchange = send(ExchangePattern.InOut, uri(NODE), new
Object(), Map.of());
+
+ assertInstanceOf(InvalidPayloadException.class,
exchange.getException());
+ assertEquals("one", data(NODE));
+ }
+
+ @Test
+ void integerBodyFailsAndKeepsTheData() throws Exception {
+ Exchange exchange = send(ExchangePattern.InOnly, uri(NODE), 42,
Map.of());
+
+ assertInstanceOf(InvalidPayloadException.class,
exchange.getException());
+ assertEquals("one", data(NODE));
+ }
+
+ @Test
+ void textBodyIsStored() throws Exception {
+ Exchange exchange = send(ExchangePattern.InOnly, uri(NODE), "two",
Map.of());
+
+ assertNull(exchange.getException());
+ assertEquals("two", data(NODE));
+ assertEquals("two", exchange.getMessage().getBody(String.class),
"InOnly leaves the body unchanged");
+ }
+
+ @Test
+ void nullBodyStoresNoData() throws Exception {
+ Exchange exchange = send(ExchangePattern.InOut, uri(NODE), null,
Map.of());
+
+ assertNull(exchange.getException());
+ assertNull(client.getData(NODE, false, null));
+ }
+
+ @Test
+ void inOnlyWriteToMissingNodeFails() throws Exception {
+ Exchange exchange = send(ExchangePattern.InOnly, uri(MISSING), "two",
Map.of());
+
+ assertInstanceOf(KeeperException.NoNodeException.class,
exchange.getException());
+ assertNull(client.exists(MISSING, false));
+ }
+
+ @Test
+ void inOnlyWriteWithStaleVersionFails() throws Exception {
+ Exchange exchange = send(ExchangePattern.InOnly, uri(NODE), "two",
Map.of(ZooKeeperMessage.ZOOKEEPER_NODE_VERSION, 7));
+
+ assertInstanceOf(KeeperException.BadVersionException.class,
exchange.getException());
+ assertEquals("one", data(NODE));
+ }
+
+ @Test
+ void inOnlyWriteWithCreateCreatesTheNodeBeforeTheExchangeCompletes()
throws Exception {
+ Exchange exchange = send(ExchangePattern.InOnly, uri(MISSING) +
"?create=true&createMode=PERSISTENT", "two", Map.of());
+
+ assertNull(exchange.getException());
+ assertEquals("two", data(MISSING));
+ }
+
+ @Test
+ void inOnlyDeleteOfMissingNodeFails() {
+ Exchange exchange = send(ExchangePattern.InOnly, uri(MISSING), "two",
delete());
+
+ assertInstanceOf(KeeperException.NoNodeException.class,
exchange.getException());
+ }
+
+ @Test
+ void deleteOfMissingNodeWithCreateDoesNotCreateIt() throws Exception {
+ Exchange exchange = send(ExchangePattern.InOut, uri(MISSING) +
"?create=true&createMode=PERSISTENT", "two", delete());
+
+ assertInstanceOf(KeeperException.NoNodeException.class,
exchange.getException());
+ assertNull(client.exists(MISSING, false), "a DELETE created the node");
+ }
+
+ @Test
+ void deleteDoesNotReadTheBody() throws Exception {
+ Exchange exchange = send(ExchangePattern.InOnly, uri(NODE),
Map.of("key", "value"), delete());
+
+ assertNull(exchange.getException());
+ assertNull(client.exists(NODE, false));
+ }
+
+ @Test
+ void listChildrenDoesNotHideAFailedWrite() throws Exception {
+ Exchange exchange = send(ExchangePattern.InOut, uri(PARENT) +
"?listChildren=true", "two",
+ Map.of(ZooKeeperMessage.ZOOKEEPER_NODE, PARENT + "/missing"));
+
+ assertInstanceOf(KeeperException.NoNodeException.class,
exchange.getException());
+ }
+
+ @Test
+ void listChildrenAfterAWrite() throws Exception {
+ Exchange exchange = send(ExchangePattern.InOut, uri(PARENT) +
"?listChildren=true", "two",
+ Map.of(ZooKeeperMessage.ZOOKEEPER_NODE, PARENT + "/child"));
+
+ assertNull(exchange.getException());
+ assertEquals(List.of("child"),
exchange.getMessage().getBody(List.class));
+ assertArrayEquals(bytes("two"), client.getData(PARENT + "/child",
false, null));
+ }
+
+ private Exchange send(ExchangePattern pattern, String uri, Object body,
Map<String, Object> headers) {
+ return template.send(uri, pattern, e -> {
+ e.getIn().setBody(body);
+ e.getIn().setHeaders(new HashMap<>(headers));
+ });
+ }
+
+ private static Map<String, Object> delete() {
+ return Map.of(ZooKeeperMessage.ZOOKEEPER_OPERATION,
ZooKeeperProducer.ZK_OPERATION_DELETE);
+ }
+
+ private String uri(String node) {
+ return "zookeeper://127.0.0.1:" + port + node;
+ }
+
+ private String data(String node) throws Exception {
+ return new String(client.getData(node, false, null),
StandardCharsets.UTF_8);
+ }
+
+ private static byte[] bytes(String text) {
+ return text.getBytes(StandardCharsets.UTF_8);
+ }
+}
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 7afa6ab7df0d..8f5992b77a54 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
@@ -2598,6 +2598,22 @@ The limit counts the whole decompressed message. For
example, when a `.tar.gz` f
`gzipDeflater` and then split into its entries, the limit applies to the whole
tar archive, not to each entry as the
limit of `tarFile` does.
+=== camel-zookeeper - the producer waits for every write and fails on a body
it cannot store
+
+The `zookeeper` producer now waits for ZooKeeper to confirm a write or a
delete of an InOnly exchange too, and fails the
+exchange when it failed. Before, an InOnly exchange was sent without waiting,
and a failure, such as a _znode_ that
+does not exist without `create=true`, a stale `CamelZooKeeperVersion` header,
or a delete of a missing _znode_, was
+only logged. An InOnly exchange still keeps its body.
+
+A body that cannot be converted to bytes, such as a `Map`, a POJO or a number,
now fails the exchange with an
+`InvalidPayloadException`. Before, the _znode_ was overwritten with no data. A
null body still stores no data.
+
+A `DELETE` of a _znode_ that does not exist no longer creates it when
`create=true` is set; it fails the exchange with
+a `NoNodeException`, as it did without `create=true`.
+
+With `listChildren=true`, a failed write or delete of an InOut exchange now
fails the exchange, instead of being
+replaced by the listing of the child _znodes_.
+
=== camel-milo - each client connection has thread pools of its own
A `milo-client` or `milo-browse` connection now handles the outcome of its
write, method call, read and browse
diff --git
a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/ZookeeperComponentBuilderFactory.java
b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/ZookeeperComponentBuilderFactory.java
index 26691a012336..9f0c2dff844f 100644
---
a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/ZookeeperComponentBuilderFactory.java
+++
b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/ZookeeperComponentBuilderFactory.java
@@ -132,6 +132,8 @@ public interface ZookeeperComponentBuilderFactory {
/**
* Should changes to the znode be 'watched' and repeatedly processed.
+ * Without it, the consumer emits the data (or the child listing) of
the
+ * znode once and does not follow later changes.
*
* The option is a: <code>boolean</code> type.
*
diff --git
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/ZooKeeperEndpointBuilderFactory.java
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/ZooKeeperEndpointBuilderFactory.java
index 3dcfa7560bfa..582e665c9c96 100644
---
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/ZooKeeperEndpointBuilderFactory.java
+++
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/ZooKeeperEndpointBuilderFactory.java
@@ -135,6 +135,8 @@ public interface ZooKeeperEndpointBuilderFactory {
}
/**
* Should changes to the znode be 'watched' and repeatedly processed.
+ * Without it, the consumer emits the data (or the child listing) of
the
+ * znode once and does not follow later changes.
*
* The option is a: <code>boolean</code> type.
*
@@ -150,6 +152,8 @@ public interface ZooKeeperEndpointBuilderFactory {
}
/**
* Should changes to the znode be 'watched' and repeatedly processed.
+ * Without it, the consumer emits the data (or the child listing) of
the
+ * znode once and does not follow later changes.
*
* The option will be converted to a <code>boolean</code> type.
*
