This is an automated email from the ASF dual-hosted git repository. acosentino pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/camel.git
commit 9f124e7149e35ea7231ea7a7dbe3b9cae0566b10 Author: Andrea Cosentino <[email protected]> AuthorDate: Thu Oct 29 19:31:42 2020 +0100 CAMEL-15770 - Kafka serialize/deserialize properties are inconsistently named - kafkaHeaderDeserializer tests --- .../apache/camel/catalog/docs/kafka-component.adoc | 4 +- .../component/kafka/KafkaComponentConfigurer.java | 10 ++-- .../component/kafka/KafkaEndpointConfigurer.java | 10 ++-- .../component/kafka/KafkaEndpointUriFactory.java | 2 +- .../org/apache/camel/component/kafka/kafka.json | 4 +- .../camel-kafka/src/main/docs/kafka-component.adoc | 4 +- .../component/kafka/KafkaConsumerFullTest.java | 2 +- .../dsl/KafkaComponentBuilderFactory.java | 30 ++++++------ .../endpoint/dsl/KafkaEndpointBuilderFactory.java | 56 +++++++++++----------- .../modules/ROOT/pages/kafka-component.adoc | 4 +- 10 files changed, 63 insertions(+), 63 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kafka-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kafka-component.adoc index 26f7fef..0e1408b 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kafka-component.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kafka-component.adoc @@ -70,8 +70,8 @@ The Kafka component supports 97 options, which are listed below. | *fetchMinBytes* (consumer) | The minimum amount of data the server should return for a fetch request. If insufficient data is available the request will wait for that much data to accumulate before answering the request. | 1 | Integer | *fetchWaitMaxMs* (consumer) | The maximum amount of time the server will block before answering the fetch request if there isn't sufficient data to immediately satisfy fetch.min.bytes | 500 | Integer | *groupId* (consumer) | A string that uniquely identifies the group of consumer processes to which this consumer belongs. By setting the same group id multiple processes indicate that they are all part of the same consumer group. This option is required for consumers. | | String +| *headerDeserializer* (consumer) | To use a custom KafkaHeaderDeserializer to deserialize kafka headers values | | KafkaHeaderDeserializer | *heartbeatIntervalMs* (consumer) | The expected time between heartbeats to the consumer coordinator when using Kafka's group management facilities. Heartbeats are used to ensure that the consumer's session stays active and to facilitate rebalancing when new consumers join or leave the group. The value must be set lower than session.timeout.ms, but typically should be set no higher than 1/3 of that value. It can be adjusted even lower to control the expected time for normal rebalances. [...] -| *kafkaHeaderDeserializer* (consumer) | To use a custom KafkaHeaderDeserializer to deserialize kafka headers values | | KafkaHeaderDeserializer | *keyDeserializer* (consumer) | Deserializer class for key that implements the Deserializer interface. | org.apache.kafka.common.serialization.StringDeserializer | String | *maxPartitionFetchBytes* (consumer) | The maximum amount of data per-partition the server will return. The maximum total memory used for a request will be #partitions max.partition.fetch.bytes. This size must be at least as large as the maximum message size the server allows or else it is possible for the producer to send messages larger than the consumer can fetch. If that happens, the consumer can get stuck trying to fetch a large message on a certain partition. | 1048576 | Integer | *maxPollIntervalMs* (consumer) | The maximum delay between invocations of poll() when using consumer group management. This places an upper bound on the amount of time that the consumer can be idle before fetching more records. If poll() is not called before expiration of this timeout, then the consumer is considered failed and the group will rebalance in order to reassign the partitions to another member. | | Long @@ -196,8 +196,8 @@ with the following path and query parameters: | *fetchMinBytes* (consumer) | The minimum amount of data the server should return for a fetch request. If insufficient data is available the request will wait for that much data to accumulate before answering the request. | 1 | Integer | *fetchWaitMaxMs* (consumer) | The maximum amount of time the server will block before answering the fetch request if there isn't sufficient data to immediately satisfy fetch.min.bytes | 500 | Integer | *groupId* (consumer) | A string that uniquely identifies the group of consumer processes to which this consumer belongs. By setting the same group id multiple processes indicate that they are all part of the same consumer group. This option is required for consumers. | | String +| *headerDeserializer* (consumer) | To use a custom KafkaHeaderDeserializer to deserialize kafka headers values | | KafkaHeaderDeserializer | *heartbeatIntervalMs* (consumer) | The expected time between heartbeats to the consumer coordinator when using Kafka's group management facilities. Heartbeats are used to ensure that the consumer's session stays active and to facilitate rebalancing when new consumers join or leave the group. The value must be set lower than session.timeout.ms, but typically should be set no higher than 1/3 of that value. It can be adjusted even lower to control the expected time for normal rebalances. [...] -| *kafkaHeaderDeserializer* (consumer) | To use a custom KafkaHeaderDeserializer to deserialize kafka headers values | | KafkaHeaderDeserializer | *keyDeserializer* (consumer) | Deserializer class for key that implements the Deserializer interface. | org.apache.kafka.common.serialization.StringDeserializer | String | *maxPartitionFetchBytes* (consumer) | The maximum amount of data per-partition the server will return. The maximum total memory used for a request will be #partitions max.partition.fetch.bytes. This size must be at least as large as the maximum message size the server allows or else it is possible for the producer to send messages larger than the consumer can fetch. If that happens, the consumer can get stuck trying to fetch a large message on a certain partition. | 1048576 | Integer | *maxPollIntervalMs* (consumer) | The maximum delay between invocations of poll() when using consumer group management. This places an upper bound on the amount of time that the consumer can be idle before fetching more records. If poll() is not called before expiration of this timeout, then the consumer is considered failed and the group will rebalance in order to reassign the partitions to another member. | | Long diff --git a/components/camel-kafka/src/generated/java/org/apache/camel/component/kafka/KafkaComponentConfigurer.java b/components/camel-kafka/src/generated/java/org/apache/camel/component/kafka/KafkaComponentConfigurer.java index c5b3a26..d154931 100644 --- a/components/camel-kafka/src/generated/java/org/apache/camel/component/kafka/KafkaComponentConfigurer.java +++ b/components/camel-kafka/src/generated/java/org/apache/camel/component/kafka/KafkaComponentConfigurer.java @@ -40,8 +40,8 @@ public class KafkaComponentConfigurer extends PropertyConfigurerSupport implemen map.put("fetchMinBytes", java.lang.Integer.class); map.put("fetchWaitMaxMs", java.lang.Integer.class); map.put("groupId", java.lang.String.class); + map.put("headerDeserializer", org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer.class); map.put("heartbeatIntervalMs", java.lang.Integer.class); - map.put("kafkaHeaderDeserializer", org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer.class); map.put("keyDeserializer", java.lang.String.class); map.put("maxPartitionFetchBytes", java.lang.Integer.class); map.put("maxPollIntervalMs", java.lang.Long.class); @@ -175,6 +175,8 @@ public class KafkaComponentConfigurer extends PropertyConfigurerSupport implemen case "fetchWaitMaxMs": getOrCreateConfiguration(target).setFetchWaitMaxMs(property(camelContext, java.lang.Integer.class, value)); return true; case "groupid": case "groupId": getOrCreateConfiguration(target).setGroupId(property(camelContext, java.lang.String.class, value)); return true; + case "headerdeserializer": + case "headerDeserializer": getOrCreateConfiguration(target).setHeaderDeserializer(property(camelContext, org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer.class, value)); return true; case "headerfilterstrategy": case "headerFilterStrategy": getOrCreateConfiguration(target).setHeaderFilterStrategy(property(camelContext, org.apache.camel.spi.HeaderFilterStrategy.class, value)); return true; case "headerserializer": @@ -183,8 +185,6 @@ public class KafkaComponentConfigurer extends PropertyConfigurerSupport implemen case "heartbeatIntervalMs": getOrCreateConfiguration(target).setHeartbeatIntervalMs(property(camelContext, java.lang.Integer.class, value)); return true; case "interceptorclasses": case "interceptorClasses": getOrCreateConfiguration(target).setInterceptorClasses(property(camelContext, java.lang.String.class, value)); return true; - case "kafkaheaderdeserializer": - case "kafkaHeaderDeserializer": getOrCreateConfiguration(target).setKafkaHeaderDeserializer(property(camelContext, org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer.class, value)); return true; case "kafkamanualcommitfactory": case "kafkaManualCommitFactory": target.setKafkaManualCommitFactory(property(camelContext, org.apache.camel.component.kafka.KafkaManualCommitFactory.class, value)); return true; case "kerberosbeforereloginmintime": @@ -377,6 +377,8 @@ public class KafkaComponentConfigurer extends PropertyConfigurerSupport implemen case "fetchWaitMaxMs": return getOrCreateConfiguration(target).getFetchWaitMaxMs(); case "groupid": case "groupId": return getOrCreateConfiguration(target).getGroupId(); + case "headerdeserializer": + case "headerDeserializer": return getOrCreateConfiguration(target).getHeaderDeserializer(); case "headerfilterstrategy": case "headerFilterStrategy": return getOrCreateConfiguration(target).getHeaderFilterStrategy(); case "headerserializer": @@ -385,8 +387,6 @@ public class KafkaComponentConfigurer extends PropertyConfigurerSupport implemen case "heartbeatIntervalMs": return getOrCreateConfiguration(target).getHeartbeatIntervalMs(); case "interceptorclasses": case "interceptorClasses": return getOrCreateConfiguration(target).getInterceptorClasses(); - case "kafkaheaderdeserializer": - case "kafkaHeaderDeserializer": return getOrCreateConfiguration(target).getKafkaHeaderDeserializer(); case "kafkamanualcommitfactory": case "kafkaManualCommitFactory": return target.getKafkaManualCommitFactory(); case "kerberosbeforereloginmintime": diff --git a/components/camel-kafka/src/generated/java/org/apache/camel/component/kafka/KafkaEndpointConfigurer.java b/components/camel-kafka/src/generated/java/org/apache/camel/component/kafka/KafkaEndpointConfigurer.java index 733a607..e85859e 100644 --- a/components/camel-kafka/src/generated/java/org/apache/camel/component/kafka/KafkaEndpointConfigurer.java +++ b/components/camel-kafka/src/generated/java/org/apache/camel/component/kafka/KafkaEndpointConfigurer.java @@ -40,8 +40,8 @@ public class KafkaEndpointConfigurer extends PropertyConfigurerSupport implement map.put("fetchMinBytes", java.lang.Integer.class); map.put("fetchWaitMaxMs", java.lang.Integer.class); map.put("groupId", java.lang.String.class); + map.put("headerDeserializer", org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer.class); map.put("heartbeatIntervalMs", java.lang.Integer.class); - map.put("kafkaHeaderDeserializer", org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer.class); map.put("keyDeserializer", java.lang.String.class); map.put("maxPartitionFetchBytes", java.lang.Integer.class); map.put("maxPollIntervalMs", java.lang.Long.class); @@ -172,6 +172,8 @@ public class KafkaEndpointConfigurer extends PropertyConfigurerSupport implement case "fetchWaitMaxMs": target.getConfiguration().setFetchWaitMaxMs(property(camelContext, java.lang.Integer.class, value)); return true; case "groupid": case "groupId": target.getConfiguration().setGroupId(property(camelContext, java.lang.String.class, value)); return true; + case "headerdeserializer": + case "headerDeserializer": target.getConfiguration().setHeaderDeserializer(property(camelContext, org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer.class, value)); return true; case "headerfilterstrategy": case "headerFilterStrategy": target.getConfiguration().setHeaderFilterStrategy(property(camelContext, org.apache.camel.spi.HeaderFilterStrategy.class, value)); return true; case "headerserializer": @@ -180,8 +182,6 @@ public class KafkaEndpointConfigurer extends PropertyConfigurerSupport implement case "heartbeatIntervalMs": target.getConfiguration().setHeartbeatIntervalMs(property(camelContext, java.lang.Integer.class, value)); return true; case "interceptorclasses": case "interceptorClasses": target.getConfiguration().setInterceptorClasses(property(camelContext, java.lang.String.class, value)); return true; - case "kafkaheaderdeserializer": - case "kafkaHeaderDeserializer": target.getConfiguration().setKafkaHeaderDeserializer(property(camelContext, org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer.class, value)); return true; case "kerberosbeforereloginmintime": case "kerberosBeforeReloginMinTime": target.getConfiguration().setKerberosBeforeReloginMinTime(property(camelContext, java.lang.Integer.class, value)); return true; case "kerberosinitcmd": @@ -374,6 +374,8 @@ public class KafkaEndpointConfigurer extends PropertyConfigurerSupport implement case "fetchWaitMaxMs": return target.getConfiguration().getFetchWaitMaxMs(); case "groupid": case "groupId": return target.getConfiguration().getGroupId(); + case "headerdeserializer": + case "headerDeserializer": return target.getConfiguration().getHeaderDeserializer(); case "headerfilterstrategy": case "headerFilterStrategy": return target.getConfiguration().getHeaderFilterStrategy(); case "headerserializer": @@ -382,8 +384,6 @@ public class KafkaEndpointConfigurer extends PropertyConfigurerSupport implement case "heartbeatIntervalMs": return target.getConfiguration().getHeartbeatIntervalMs(); case "interceptorclasses": case "interceptorClasses": return target.getConfiguration().getInterceptorClasses(); - case "kafkaheaderdeserializer": - case "kafkaHeaderDeserializer": return target.getConfiguration().getKafkaHeaderDeserializer(); case "kerberosbeforereloginmintime": case "kerberosBeforeReloginMinTime": return target.getConfiguration().getKerberosBeforeReloginMinTime(); case "kerberosinitcmd": diff --git a/components/camel-kafka/src/generated/java/org/apache/camel/component/kafka/KafkaEndpointUriFactory.java b/components/camel-kafka/src/generated/java/org/apache/camel/component/kafka/KafkaEndpointUriFactory.java index e815bdc..92cbe3e 100644 --- a/components/camel-kafka/src/generated/java/org/apache/camel/component/kafka/KafkaEndpointUriFactory.java +++ b/components/camel-kafka/src/generated/java/org/apache/camel/component/kafka/KafkaEndpointUriFactory.java @@ -109,13 +109,13 @@ public class KafkaEndpointUriFactory extends org.apache.camel.support.component. props.add("sslEnabledProtocols"); props.add("sslKeystoreLocation"); props.add("schemaRegistryURL"); + props.add("headerDeserializer"); props.add("maxInFlightRequest"); props.add("exchangePattern"); props.add("valueSerializer"); props.add("autoOffsetReset"); props.add("seekTo"); props.add("requestTimeoutMs"); - props.add("kafkaHeaderDeserializer"); props.add("fetchMaxBytes"); props.add("checkCrcs"); props.add("partitionAssignor"); diff --git a/components/camel-kafka/src/generated/resources/org/apache/camel/component/kafka/kafka.json b/components/camel-kafka/src/generated/resources/org/apache/camel/component/kafka/kafka.json index 319e694..2cbc60d 100644 --- a/components/camel-kafka/src/generated/resources/org/apache/camel/component/kafka/kafka.json +++ b/components/camel-kafka/src/generated/resources/org/apache/camel/component/kafka/kafka.json @@ -44,8 +44,8 @@ "fetchMinBytes": { "kind": "property", "displayName": "Fetch Min Bytes", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "secret": false, "defaultValue": "1", "configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration", "configurationField": "configuration", "description": "The minimum amount of data the server should return for a fetch request. If insufficient data is available the [...] "fetchWaitMaxMs": { "kind": "property", "displayName": "Fetch Wait Max Ms", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "secret": false, "defaultValue": "500", "configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration", "configurationField": "configuration", "description": "The maximum amount of time the server will block before answering the fetch request if there isn't suffici [...] "groupId": { "kind": "property", "displayName": "Group Id", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration", "configurationField": "configuration", "description": "A string that uniquely identifies the group of consumer processes to which this consumer belongs. By setting the same group id multiple processes i [...] + "headerDeserializer": { "kind": "property", "displayName": "Header Deserializer", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration", "configurationField": "configuration", "description": "To use a custom KafkaHeaderDeserializer to deserialize kafka headers values" }, "heartbeatIntervalMs": { "kind": "property", "displayName": "Heartbeat Interval Ms", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "secret": false, "defaultValue": "3000", "configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration", "configurationField": "configuration", "description": "The expected time between heartbeats to the consumer coordinator when using Kafka's group manage [...] - "kafkaHeaderDeserializer": { "kind": "property", "displayName": "Kafka Header Deserializer", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration", "configurationField": "configuration", "description": "To use a custom KafkaHeaderDeserializer to deserialize kafka headers [...] "keyDeserializer": { "kind": "property", "displayName": "Key Deserializer", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "org.apache.kafka.common.serialization.StringDeserializer", "configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration", "configurationField": "configuration", "description": "Deserializer class for key that implements the Deserial [...] "maxPartitionFetchBytes": { "kind": "property", "displayName": "Max Partition Fetch Bytes", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "secret": false, "defaultValue": "1048576", "configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration", "configurationField": "configuration", "description": "The maximum amount of data per-partition the server will return. The maximum total mem [...] "maxPollIntervalMs": { "kind": "property", "displayName": "Max Poll Interval Ms", "group": "consumer", "label": "consumer", "required": false, "type": "duration", "javaType": "java.lang.Long", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration", "configurationField": "configuration", "description": "The maximum delay between invocations of poll() when using consumer group management. This places an upper bound on the amoun [...] @@ -143,8 +143,8 @@ "fetchMinBytes": { "kind": "parameter", "displayName": "Fetch Min Bytes", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "secret": false, "defaultValue": "1", "configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration", "configurationField": "configuration", "description": "The minimum amount of data the server should return for a fetch request. If insufficient data is available the [...] "fetchWaitMaxMs": { "kind": "parameter", "displayName": "Fetch Wait Max Ms", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "secret": false, "defaultValue": "500", "configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration", "configurationField": "configuration", "description": "The maximum amount of time the server will block before answering the fetch request if there isn't suffic [...] "groupId": { "kind": "parameter", "displayName": "Group Id", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration", "configurationField": "configuration", "description": "A string that uniquely identifies the group of consumer processes to which this consumer belongs. By setting the same group id multiple processes [...] + "headerDeserializer": { "kind": "parameter", "displayName": "Header Deserializer", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration", "configurationField": "configuration", "description": "To use a custom KafkaHeaderDeserializer to deserialize kafka headers values" }, "heartbeatIntervalMs": { "kind": "parameter", "displayName": "Heartbeat Interval Ms", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "secret": false, "defaultValue": "3000", "configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration", "configurationField": "configuration", "description": "The expected time between heartbeats to the consumer coordinator when using Kafka's group manag [...] - "kafkaHeaderDeserializer": { "kind": "parameter", "displayName": "Kafka Header Deserializer", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration", "configurationField": "configuration", "description": "To use a custom KafkaHeaderDeserializer to deserialize kafka header [...] "keyDeserializer": { "kind": "parameter", "displayName": "Key Deserializer", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "org.apache.kafka.common.serialization.StringDeserializer", "configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration", "configurationField": "configuration", "description": "Deserializer class for key that implements the Deseria [...] "maxPartitionFetchBytes": { "kind": "parameter", "displayName": "Max Partition Fetch Bytes", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "secret": false, "defaultValue": "1048576", "configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration", "configurationField": "configuration", "description": "The maximum amount of data per-partition the server will return. The maximum total me [...] "maxPollIntervalMs": { "kind": "parameter", "displayName": "Max Poll Interval Ms", "group": "consumer", "label": "consumer", "required": false, "type": "duration", "javaType": "java.lang.Long", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration", "configurationField": "configuration", "description": "The maximum delay between invocations of poll() when using consumer group management. This places an upper bound on the amou [...] diff --git a/components/camel-kafka/src/main/docs/kafka-component.adoc b/components/camel-kafka/src/main/docs/kafka-component.adoc index 26f7fef..0e1408b 100644 --- a/components/camel-kafka/src/main/docs/kafka-component.adoc +++ b/components/camel-kafka/src/main/docs/kafka-component.adoc @@ -70,8 +70,8 @@ The Kafka component supports 97 options, which are listed below. | *fetchMinBytes* (consumer) | The minimum amount of data the server should return for a fetch request. If insufficient data is available the request will wait for that much data to accumulate before answering the request. | 1 | Integer | *fetchWaitMaxMs* (consumer) | The maximum amount of time the server will block before answering the fetch request if there isn't sufficient data to immediately satisfy fetch.min.bytes | 500 | Integer | *groupId* (consumer) | A string that uniquely identifies the group of consumer processes to which this consumer belongs. By setting the same group id multiple processes indicate that they are all part of the same consumer group. This option is required for consumers. | | String +| *headerDeserializer* (consumer) | To use a custom KafkaHeaderDeserializer to deserialize kafka headers values | | KafkaHeaderDeserializer | *heartbeatIntervalMs* (consumer) | The expected time between heartbeats to the consumer coordinator when using Kafka's group management facilities. Heartbeats are used to ensure that the consumer's session stays active and to facilitate rebalancing when new consumers join or leave the group. The value must be set lower than session.timeout.ms, but typically should be set no higher than 1/3 of that value. It can be adjusted even lower to control the expected time for normal rebalances. [...] -| *kafkaHeaderDeserializer* (consumer) | To use a custom KafkaHeaderDeserializer to deserialize kafka headers values | | KafkaHeaderDeserializer | *keyDeserializer* (consumer) | Deserializer class for key that implements the Deserializer interface. | org.apache.kafka.common.serialization.StringDeserializer | String | *maxPartitionFetchBytes* (consumer) | The maximum amount of data per-partition the server will return. The maximum total memory used for a request will be #partitions max.partition.fetch.bytes. This size must be at least as large as the maximum message size the server allows or else it is possible for the producer to send messages larger than the consumer can fetch. If that happens, the consumer can get stuck trying to fetch a large message on a certain partition. | 1048576 | Integer | *maxPollIntervalMs* (consumer) | The maximum delay between invocations of poll() when using consumer group management. This places an upper bound on the amount of time that the consumer can be idle before fetching more records. If poll() is not called before expiration of this timeout, then the consumer is considered failed and the group will rebalance in order to reassign the partitions to another member. | | Long @@ -196,8 +196,8 @@ with the following path and query parameters: | *fetchMinBytes* (consumer) | The minimum amount of data the server should return for a fetch request. If insufficient data is available the request will wait for that much data to accumulate before answering the request. | 1 | Integer | *fetchWaitMaxMs* (consumer) | The maximum amount of time the server will block before answering the fetch request if there isn't sufficient data to immediately satisfy fetch.min.bytes | 500 | Integer | *groupId* (consumer) | A string that uniquely identifies the group of consumer processes to which this consumer belongs. By setting the same group id multiple processes indicate that they are all part of the same consumer group. This option is required for consumers. | | String +| *headerDeserializer* (consumer) | To use a custom KafkaHeaderDeserializer to deserialize kafka headers values | | KafkaHeaderDeserializer | *heartbeatIntervalMs* (consumer) | The expected time between heartbeats to the consumer coordinator when using Kafka's group management facilities. Heartbeats are used to ensure that the consumer's session stays active and to facilitate rebalancing when new consumers join or leave the group. The value must be set lower than session.timeout.ms, but typically should be set no higher than 1/3 of that value. It can be adjusted even lower to control the expected time for normal rebalances. [...] -| *kafkaHeaderDeserializer* (consumer) | To use a custom KafkaHeaderDeserializer to deserialize kafka headers values | | KafkaHeaderDeserializer | *keyDeserializer* (consumer) | Deserializer class for key that implements the Deserializer interface. | org.apache.kafka.common.serialization.StringDeserializer | String | *maxPartitionFetchBytes* (consumer) | The maximum amount of data per-partition the server will return. The maximum total memory used for a request will be #partitions max.partition.fetch.bytes. This size must be at least as large as the maximum message size the server allows or else it is possible for the producer to send messages larger than the consumer can fetch. If that happens, the consumer can get stuck trying to fetch a large message on a certain partition. | 1048576 | Integer | *maxPollIntervalMs* (consumer) | The maximum delay between invocations of poll() when using consumer group management. This places an upper bound on the amount of time that the consumer can be idle before fetching more records. If poll() is not called before expiration of this timeout, then the consumer is considered failed and the group will rebalance in order to reassign the partitions to another member. | | Long diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerFullTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerFullTest.java index 850cefa..8e1cad7 100644 --- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerFullTest.java +++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerFullTest.java @@ -195,7 +195,7 @@ public class KafkaConsumerFullTest extends BaseEmbeddedKafkaTest { public void headerDeserializerCouldBeOverridden() { KafkaEndpoint kafkaEndpoint = context.getEndpoint("kafka:random_topic?kafkaHeaderDeserializer=#myHeaderDeserializer", KafkaEndpoint.class); - assertIsInstanceOf(MyKafkaHeaderDeserializer.class, kafkaEndpoint.getConfiguration().getKafkaHeaderDeserializer()); + assertIsInstanceOf(MyKafkaHeaderDeserializer.class, kafkaEndpoint.getConfiguration().getHeaderDeserializer()); } private static class MyKafkaHeaderDeserializer extends DefaultKafkaHeaderDeserializer { diff --git a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/KafkaComponentBuilderFactory.java b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/KafkaComponentBuilderFactory.java index ee66c62..6a526bf 100644 --- a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/KafkaComponentBuilderFactory.java +++ b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/KafkaComponentBuilderFactory.java @@ -388,6 +388,20 @@ public interface KafkaComponentBuilderFactory { return this; } /** + * To use a custom KafkaHeaderDeserializer to deserialize kafka headers + * values. + * + * The option is a: + * <code>org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer</code> type. + * + * Group: consumer + */ + default KafkaComponentBuilder headerDeserializer( + org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer headerDeserializer) { + doSetProperty("headerDeserializer", headerDeserializer); + return this; + } + /** * The expected time between heartbeats to the consumer coordinator when * using Kafka's group management facilities. Heartbeats are used to * ensure that the consumer's session stays active and to facilitate @@ -407,20 +421,6 @@ public interface KafkaComponentBuilderFactory { return this; } /** - * To use a custom KafkaHeaderDeserializer to deserialize kafka headers - * values. - * - * The option is a: - * <code>org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer</code> type. - * - * Group: consumer - */ - default KafkaComponentBuilder kafkaHeaderDeserializer( - org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer kafkaHeaderDeserializer) { - doSetProperty("kafkaHeaderDeserializer", kafkaHeaderDeserializer); - return this; - } - /** * Deserializer class for key that implements the Deserializer * interface. * @@ -1587,8 +1587,8 @@ public interface KafkaComponentBuilderFactory { case "fetchMinBytes": getOrCreateConfiguration((KafkaComponent) component).setFetchMinBytes((java.lang.Integer) value); return true; case "fetchWaitMaxMs": getOrCreateConfiguration((KafkaComponent) component).setFetchWaitMaxMs((java.lang.Integer) value); return true; case "groupId": getOrCreateConfiguration((KafkaComponent) component).setGroupId((java.lang.String) value); return true; + case "headerDeserializer": getOrCreateConfiguration((KafkaComponent) component).setHeaderDeserializer((org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer) value); return true; case "heartbeatIntervalMs": getOrCreateConfiguration((KafkaComponent) component).setHeartbeatIntervalMs((java.lang.Integer) value); return true; - case "kafkaHeaderDeserializer": getOrCreateConfiguration((KafkaComponent) component).setKafkaHeaderDeserializer((org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer) value); return true; case "keyDeserializer": getOrCreateConfiguration((KafkaComponent) component).setKeyDeserializer((java.lang.String) value); return true; case "maxPartitionFetchBytes": getOrCreateConfiguration((KafkaComponent) component).setMaxPartitionFetchBytes((java.lang.Integer) value); return true; case "maxPollIntervalMs": getOrCreateConfiguration((KafkaComponent) component).setMaxPollIntervalMs((java.lang.Long) value); return true; diff --git a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/KafkaEndpointBuilderFactory.java b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/KafkaEndpointBuilderFactory.java index 823068b..9921b04 100644 --- a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/KafkaEndpointBuilderFactory.java +++ b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/KafkaEndpointBuilderFactory.java @@ -638,6 +638,34 @@ public interface KafkaEndpointBuilderFactory { return this; } /** + * To use a custom KafkaHeaderDeserializer to deserialize kafka headers + * values. + * + * The option is a: + * <code>org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer</code> type. + * + * Group: consumer + */ + default KafkaEndpointConsumerBuilder headerDeserializer( + Object headerDeserializer) { + doSetProperty("headerDeserializer", headerDeserializer); + return this; + } + /** + * To use a custom KafkaHeaderDeserializer to deserialize kafka headers + * values. + * + * The option will be converted to a + * <code>org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer</code> type. + * + * Group: consumer + */ + default KafkaEndpointConsumerBuilder headerDeserializer( + String headerDeserializer) { + doSetProperty("headerDeserializer", headerDeserializer); + return this; + } + /** * The expected time between heartbeats to the consumer coordinator when * using Kafka's group management facilities. Heartbeats are used to * ensure that the consumer's session stays active and to facilitate @@ -677,34 +705,6 @@ public interface KafkaEndpointBuilderFactory { return this; } /** - * To use a custom KafkaHeaderDeserializer to deserialize kafka headers - * values. - * - * The option is a: - * <code>org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer</code> type. - * - * Group: consumer - */ - default KafkaEndpointConsumerBuilder kafkaHeaderDeserializer( - Object kafkaHeaderDeserializer) { - doSetProperty("kafkaHeaderDeserializer", kafkaHeaderDeserializer); - return this; - } - /** - * To use a custom KafkaHeaderDeserializer to deserialize kafka headers - * values. - * - * The option will be converted to a - * <code>org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer</code> type. - * - * Group: consumer - */ - default KafkaEndpointConsumerBuilder kafkaHeaderDeserializer( - String kafkaHeaderDeserializer) { - doSetProperty("kafkaHeaderDeserializer", kafkaHeaderDeserializer); - return this; - } - /** * Deserializer class for key that implements the Deserializer * interface. * diff --git a/docs/components/modules/ROOT/pages/kafka-component.adoc b/docs/components/modules/ROOT/pages/kafka-component.adoc index 9280343..c39b2da 100644 --- a/docs/components/modules/ROOT/pages/kafka-component.adoc +++ b/docs/components/modules/ROOT/pages/kafka-component.adoc @@ -72,8 +72,8 @@ The Kafka component supports 97 options, which are listed below. | *fetchMinBytes* (consumer) | The minimum amount of data the server should return for a fetch request. If insufficient data is available the request will wait for that much data to accumulate before answering the request. | 1 | Integer | *fetchWaitMaxMs* (consumer) | The maximum amount of time the server will block before answering the fetch request if there isn't sufficient data to immediately satisfy fetch.min.bytes | 500 | Integer | *groupId* (consumer) | A string that uniquely identifies the group of consumer processes to which this consumer belongs. By setting the same group id multiple processes indicate that they are all part of the same consumer group. This option is required for consumers. | | String +| *headerDeserializer* (consumer) | To use a custom KafkaHeaderDeserializer to deserialize kafka headers values | | KafkaHeaderDeserializer | *heartbeatIntervalMs* (consumer) | The expected time between heartbeats to the consumer coordinator when using Kafka's group management facilities. Heartbeats are used to ensure that the consumer's session stays active and to facilitate rebalancing when new consumers join or leave the group. The value must be set lower than session.timeout.ms, but typically should be set no higher than 1/3 of that value. It can be adjusted even lower to control the expected time for normal rebalances. [...] -| *kafkaHeaderDeserializer* (consumer) | To use a custom KafkaHeaderDeserializer to deserialize kafka headers values | | KafkaHeaderDeserializer | *keyDeserializer* (consumer) | Deserializer class for key that implements the Deserializer interface. | org.apache.kafka.common.serialization.StringDeserializer | String | *maxPartitionFetchBytes* (consumer) | The maximum amount of data per-partition the server will return. The maximum total memory used for a request will be #partitions max.partition.fetch.bytes. This size must be at least as large as the maximum message size the server allows or else it is possible for the producer to send messages larger than the consumer can fetch. If that happens, the consumer can get stuck trying to fetch a large message on a certain partition. | 1048576 | Integer | *maxPollIntervalMs* (consumer) | The maximum delay between invocations of poll() when using consumer group management. This places an upper bound on the amount of time that the consumer can be idle before fetching more records. If poll() is not called before expiration of this timeout, then the consumer is considered failed and the group will rebalance in order to reassign the partitions to another member. | | Long @@ -198,8 +198,8 @@ with the following path and query parameters: | *fetchMinBytes* (consumer) | The minimum amount of data the server should return for a fetch request. If insufficient data is available the request will wait for that much data to accumulate before answering the request. | 1 | Integer | *fetchWaitMaxMs* (consumer) | The maximum amount of time the server will block before answering the fetch request if there isn't sufficient data to immediately satisfy fetch.min.bytes | 500 | Integer | *groupId* (consumer) | A string that uniquely identifies the group of consumer processes to which this consumer belongs. By setting the same group id multiple processes indicate that they are all part of the same consumer group. This option is required for consumers. | | String +| *headerDeserializer* (consumer) | To use a custom KafkaHeaderDeserializer to deserialize kafka headers values | | KafkaHeaderDeserializer | *heartbeatIntervalMs* (consumer) | The expected time between heartbeats to the consumer coordinator when using Kafka's group management facilities. Heartbeats are used to ensure that the consumer's session stays active and to facilitate rebalancing when new consumers join or leave the group. The value must be set lower than session.timeout.ms, but typically should be set no higher than 1/3 of that value. It can be adjusted even lower to control the expected time for normal rebalances. [...] -| *kafkaHeaderDeserializer* (consumer) | To use a custom KafkaHeaderDeserializer to deserialize kafka headers values | | KafkaHeaderDeserializer | *keyDeserializer* (consumer) | Deserializer class for key that implements the Deserializer interface. | org.apache.kafka.common.serialization.StringDeserializer | String | *maxPartitionFetchBytes* (consumer) | The maximum amount of data per-partition the server will return. The maximum total memory used for a request will be #partitions max.partition.fetch.bytes. This size must be at least as large as the maximum message size the server allows or else it is possible for the producer to send messages larger than the consumer can fetch. If that happens, the consumer can get stuck trying to fetch a large message on a certain partition. | 1048576 | Integer | *maxPollIntervalMs* (consumer) | The maximum delay between invocations of poll() when using consumer group management. This places an upper bound on the amount of time that the consumer can be idle before fetching more records. If poll() is not called before expiration of this timeout, then the consumer is considered failed and the group will rebalance in order to reassign the partitions to another member. | | Long
