This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26194-zookeeper-producer in repository https://gitbox.apache.org/repos/asf/camel.git
commit 9108e5129ebd86eff37dbf846ea2527cb690534d Author: Claus Ibsen <[email protected]> AuthorDate: Tue Nov 3 00:07:31 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 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 10f5a47c1b89..71c97b7fdb9b 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 @@ -2360,6 +2360,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. *
