This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-spring-boot.git
The following commit(s) were added to refs/heads/main by this push:
new b02549e32d2 Regen
b02549e32d2 is described below
commit b02549e32d2f3703e9f011a6e90715a114f80d7d
Author: Claus Ibsen <claus.ib...@gmail.com>
AuthorDate: Mon Jan 2 11:59:36 2023 +0100
Regen
---
.../camel/springboot/catalog/components/kafka.json | 12 +++---
.../camel-kafka-starter/src/main/docs/kafka.json | 13 +++---
.../springboot/KafkaComponentConfiguration.java | 50 +++++++++++++++-------
3 files changed, 46 insertions(+), 29 deletions(-)
diff --git
a/catalog/camel-catalog-provider-springboot/src/main/resources/org/apache/camel/springboot/catalog/components/kafka.json
b/catalog/camel-catalog-provider-springboot/src/main/resources/org/apache/camel/springboot/catalog/components/kafka.json
index 6afd6d5d6a7..880ea3d5298 100644
---
a/catalog/camel-catalog-provider-springboot/src/main/resources/org/apache/camel/springboot/catalog/components/kafka.json
+++
b/catalog/camel-catalog-provider-springboot/src/main/resources/org/apache/camel/springboot/catalog/components/kafka.json
@@ -71,7 +71,7 @@
"compressionCodec": { "kind": "property", "displayName": "Compression
Codec", "group": "producer", "label": "producer", "required": false, "type":
"string", "javaType": "java.lang.String", "enum": [ "none", "gzip", "snappy",
"lz4", "zstd" ], "deprecated": false, "autowired": false, "secret": false,
"defaultValue": "none", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "This parameter allows you to spe [...]
"connectionMaxIdleMs": { "kind": "property", "displayName": "Connection
Max Idle Ms", "group": "producer", "label": "producer", "required": false,
"type": "integer", "javaType": "java.lang.Integer", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": "540000",
"configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration",
"configurationField": "configuration", "description": "Close idle connections
after the number of milliseconds specified by this [...]
"deliveryTimeoutMs": { "kind": "property", "displayName": "Delivery
Timeout Ms", "group": "producer", "label": "producer", "required": false,
"type": "integer", "javaType": "java.lang.Integer", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": "120000",
"configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration",
"configurationField": "configuration", "description": "An upper bound on the
time to report success or failure after a call to send() [...]
- "enableIdempotence": { "kind": "property", "displayName": "Enable
Idempotence", "group": "producer", "label": "producer", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "If set to 'true' the producer will ensure that
exactly one copy of each message is written i [...]
+ "enableIdempotence": { "kind": "property", "displayName": "Enable
Idempotence", "group": "producer", "label": "producer", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": true, "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "When set to 'true', the producer will ensure
that exactly one copy of each message is written [...]
"headerSerializer": { "kind": "property", "displayName": "Header
Serializer", "group": "producer", "label": "producer", "required": false,
"type": "object", "javaType":
"org.apache.camel.component.kafka.serde.KafkaHeaderSerializer", "deprecated":
false, "autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "To use a custom KafkaHeaderSerializer to
serialize kafka headers [...]
"key": { "kind": "property", "displayName": "Key", "group": "producer",
"label": "producer", "required": false, "type": "string", "javaType":
"java.lang.String", "deprecated": false, "autowired": false, "secret": false,
"configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration",
"configurationField": "configuration", "description": "The record key (or null
if no key is specified). If this option has been configured then it take
precedence over header KafkaConstants#KEY" },
"keySerializer": { "kind": "property", "displayName": "Key Serializer",
"group": "producer", "label": "producer", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "defaultValue":
"org.apache.kafka.common.serialization.StringSerializer", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "The serializer class for keys (defaults t [...]
@@ -91,9 +91,9 @@
"receiveBufferBytes": { "kind": "property", "displayName": "Receive Buffer
Bytes", "group": "producer", "label": "producer", "required": false, "type":
"integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": "65536", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "The size of the TCP receive buffer (SO_RCVBUF)
to use when reading data." },
"reconnectBackoffMs": { "kind": "property", "displayName": "Reconnect
Backoff Ms", "group": "producer", "label": "producer", "required": false,
"type": "integer", "javaType": "java.lang.Integer", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": "50",
"configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration",
"configurationField": "configuration", "description": "The amount of time to
wait before attempting to reconnect to a given host. This [...]
"recordMetadata": { "kind": "property", "displayName": "Record Metadata",
"group": "producer", "label": "producer", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": true, "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "Whether the producer should store the
RecordMetadata results from sending to Kafka. The results are [...]
- "requestRequiredAcks": { "kind": "property", "displayName": "Request
Required Acks", "group": "producer", "label": "producer", "required": false,
"type": "string", "javaType": "java.lang.String", "enum": [ "-1", "0", "1",
"all" ], "deprecated": false, "autowired": false, "secret": false,
"defaultValue": "1", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "The number of acknowledgments the producer req
[...]
+ "requestRequiredAcks": { "kind": "property", "displayName": "Request
Required Acks", "group": "producer", "label": "producer", "required": false,
"type": "string", "javaType": "java.lang.String", "enum": [ "all", "-1", "0",
"1" ], "deprecated": false, "autowired": false, "secret": false,
"defaultValue": "all", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "The number of acknowledgments the producer r
[...]
"requestTimeoutMs": { "kind": "property", "displayName": "Request Timeout
Ms", "group": "producer", "label": "producer", "required": false, "type":
"integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": "30000", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "The amount of time the broker will wait trying
to meet the request.required.acks [...]
- "retries": { "kind": "property", "displayName": "Retries", "group":
"producer", "label": "producer", "required": false, "type": "integer",
"javaType": "java.lang.Integer", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": "0", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "Setting a value greater than zero will cause
the client to resend any record whose send fails with a pote [...]
+ "retries": { "kind": "property", "displayName": "Retries", "group":
"producer", "label": "producer", "required": false, "type": "integer",
"javaType": "java.lang.Integer", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "Setting a value greater than zero will cause
the client to resend any record whose send fails with a potentially transient
err [...]
"retryBackoffMs": { "kind": "property", "displayName": "Retry Backoff Ms",
"group": "producer", "label": "producer", "required": false, "type": "integer",
"javaType": "java.lang.Integer", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": "100", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "Before each retry, the producer refreshes the
metadata of relevant topics to see if a n [...]
"sendBufferBytes": { "kind": "property", "displayName": "Send Buffer
Bytes", "group": "producer", "label": "producer", "required": false, "type":
"integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": "131072", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "Socket write buffer size" },
"valueSerializer": { "kind": "property", "displayName": "Value
Serializer", "group": "producer", "label": "producer", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "defaultValue":
"org.apache.kafka.common.serialization.StringSerializer", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "The serializer class for messages." },
@@ -193,7 +193,7 @@
"compressionCodec": { "kind": "parameter", "displayName": "Compression
Codec", "group": "producer", "label": "producer", "required": false, "type":
"string", "javaType": "java.lang.String", "enum": [ "none", "gzip", "snappy",
"lz4", "zstd" ], "deprecated": false, "autowired": false, "secret": false,
"defaultValue": "none", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "This parameter allows you to sp [...]
"connectionMaxIdleMs": { "kind": "parameter", "displayName": "Connection
Max Idle Ms", "group": "producer", "label": "producer", "required": false,
"type": "integer", "javaType": "java.lang.Integer", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": "540000",
"configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration",
"configurationField": "configuration", "description": "Close idle connections
after the number of milliseconds specified by thi [...]
"deliveryTimeoutMs": { "kind": "parameter", "displayName": "Delivery
Timeout Ms", "group": "producer", "label": "producer", "required": false,
"type": "integer", "javaType": "java.lang.Integer", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": "120000",
"configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration",
"configurationField": "configuration", "description": "An upper bound on the
time to report success or failure after a call to send( [...]
- "enableIdempotence": { "kind": "parameter", "displayName": "Enable
Idempotence", "group": "producer", "label": "producer", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "If set to 'true' the producer will ensure that
exactly one copy of each message is written [...]
+ "enableIdempotence": { "kind": "parameter", "displayName": "Enable
Idempotence", "group": "producer", "label": "producer", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": true, "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "When set to 'true', the producer will ensure
that exactly one copy of each message is writte [...]
"headerSerializer": { "kind": "parameter", "displayName": "Header
Serializer", "group": "producer", "label": "producer", "required": false,
"type": "object", "javaType":
"org.apache.camel.component.kafka.serde.KafkaHeaderSerializer", "deprecated":
false, "autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "To use a custom KafkaHeaderSerializer to
serialize kafka headers [...]
"key": { "kind": "parameter", "displayName": "Key", "group": "producer",
"label": "producer", "required": false, "type": "string", "javaType":
"java.lang.String", "deprecated": false, "autowired": false, "secret": false,
"configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration",
"configurationField": "configuration", "description": "The record key (or null
if no key is specified). If this option has been configured then it take
precedence over header KafkaConstants#KEY" },
"keySerializer": { "kind": "parameter", "displayName": "Key Serializer",
"group": "producer", "label": "producer", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "defaultValue":
"org.apache.kafka.common.serialization.StringSerializer", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "The serializer class for keys (defaults [...]
@@ -212,9 +212,9 @@
"receiveBufferBytes": { "kind": "parameter", "displayName": "Receive
Buffer Bytes", "group": "producer", "label": "producer", "required": false,
"type": "integer", "javaType": "java.lang.Integer", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": "65536",
"configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration",
"configurationField": "configuration", "description": "The size of the TCP
receive buffer (SO_RCVBUF) to use when reading data." },
"reconnectBackoffMs": { "kind": "parameter", "displayName": "Reconnect
Backoff Ms", "group": "producer", "label": "producer", "required": false,
"type": "integer", "javaType": "java.lang.Integer", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": "50",
"configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration",
"configurationField": "configuration", "description": "The amount of time to
wait before attempting to reconnect to a given host. This [...]
"recordMetadata": { "kind": "parameter", "displayName": "Record Metadata",
"group": "producer", "label": "producer", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": true, "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "Whether the producer should store the
RecordMetadata results from sending to Kafka. The results ar [...]
- "requestRequiredAcks": { "kind": "parameter", "displayName": "Request
Required Acks", "group": "producer", "label": "producer", "required": false,
"type": "string", "javaType": "java.lang.String", "enum": [ "-1", "0", "1",
"all" ], "deprecated": false, "autowired": false, "secret": false,
"defaultValue": "1", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "The number of acknowledgments the producer re
[...]
+ "requestRequiredAcks": { "kind": "parameter", "displayName": "Request
Required Acks", "group": "producer", "label": "producer", "required": false,
"type": "string", "javaType": "java.lang.String", "enum": [ "all", "-1", "0",
"1" ], "deprecated": false, "autowired": false, "secret": false,
"defaultValue": "all", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "The number of acknowledgments the producer
[...]
"requestTimeoutMs": { "kind": "parameter", "displayName": "Request Timeout
Ms", "group": "producer", "label": "producer", "required": false, "type":
"integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": "30000", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "The amount of time the broker will wait trying
to meet the request.required.acks [...]
- "retries": { "kind": "parameter", "displayName": "Retries", "group":
"producer", "label": "producer", "required": false, "type": "integer",
"javaType": "java.lang.Integer", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": "0", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "Setting a value greater than zero will cause
the client to resend any record whose send fails with a pot [...]
+ "retries": { "kind": "parameter", "displayName": "Retries", "group":
"producer", "label": "producer", "required": false, "type": "integer",
"javaType": "java.lang.Integer", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "Setting a value greater than zero will cause
the client to resend any record whose send fails with a potentially transient
er [...]
"retryBackoffMs": { "kind": "parameter", "displayName": "Retry Backoff
Ms", "group": "producer", "label": "producer", "required": false, "type":
"integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": "100", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "Before each retry, the producer refreshes the
metadata of relevant topics to see if a [...]
"sendBufferBytes": { "kind": "parameter", "displayName": "Send Buffer
Bytes", "group": "producer", "label": "producer", "required": false, "type":
"integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": "131072", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "Socket write buffer size" },
"valueSerializer": { "kind": "parameter", "displayName": "Value
Serializer", "group": "producer", "label": "producer", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "defaultValue":
"org.apache.kafka.common.serialization.StringSerializer", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "The serializer class for messages." },
diff --git a/components-starter/camel-kafka-starter/src/main/docs/kafka.json
b/components-starter/camel-kafka-starter/src/main/docs/kafka.json
index 27fa395c3d4..4b17296efda 100644
--- a/components-starter/camel-kafka-starter/src/main/docs/kafka.json
+++ b/components-starter/camel-kafka-starter/src/main/docs/kafka.json
@@ -170,9 +170,9 @@
{
"name": "camel.component.kafka.enable-idempotence",
"type": "java.lang.Boolean",
- "description": "If set to 'true' the producer will ensure that exactly
one copy of each message is written in the stream. If 'false', producer retries
may write duplicates of the retried message in the stream. If set to true this
option will require max.in.flight.requests.per.connection to be set to 1 and
retries cannot be zero and additionally acks must be set to 'all'.",
+ "description": "When set to 'true', the producer will ensure that
exactly one copy of each message is written in the stream. If 'false', producer
retries due to broker failures, etc., may write duplicates of the retried
message in the stream. Note that enabling idempotence requires
max.in.flight.requests.per.connection to be less than or equal to 5 (with
message ordering preserved for any allowable value), retries to be greater than
0, and acks must be 'all'. Idempotence is enabled [...]
"sourceType":
"org.apache.camel.component.kafka.springboot.KafkaComponentConfiguration",
- "defaultValue": false
+ "defaultValue": true
},
{
"name": "camel.component.kafka.enabled",
@@ -488,9 +488,9 @@
{
"name": "camel.component.kafka.request-required-acks",
"type": "java.lang.String",
- "description": "The number of acknowledgments the producer requires the
leader to have received before considering a request complete. This controls
the durability of records that are sent. The following settings are common:
acks=0 If set to zero then the producer will not wait for any acknowledgment
from the server at all. The record will be immediately added to the socket
buffer and considered sent. No guarantee can be made that the server has
received the record in this case, an [...]
+ "description": "The number of acknowledgments the producer requires the
leader to have received before considering a request complete. This controls
the durability of records that are sent. The following settings are allowed:
acks=0 If set to zero then the producer will not wait for any acknowledgment
from the server at all. The record will be immediately added to the socket
buffer and considered sent. No guarantee can be made that the server has
received the record in this case, a [...]
"sourceType":
"org.apache.camel.component.kafka.springboot.KafkaComponentConfiguration",
- "defaultValue": "1"
+ "defaultValue": "all"
},
{
"name": "camel.component.kafka.request-timeout-ms",
@@ -502,9 +502,8 @@
{
"name": "camel.component.kafka.retries",
"type": "java.lang.Integer",
- "description": "Setting a value greater than zero will cause the client
to resend any record whose send fails with a potentially transient error. Note
that this retry is no different than if the client resent the record upon
receiving the error. Allowing retries will potentially change the ordering of
records because if two records are sent to a single partition, and the first
fails and is retried but the second succeeds, then the second record may appear
first.",
- "sourceType":
"org.apache.camel.component.kafka.springboot.KafkaComponentConfiguration",
- "defaultValue": 0
+ "description": "Setting a value greater than zero will cause the client
to resend any record whose send fails with a potentially transient error. Note
that this retry is no different than if the client resent the record upon
receiving the error. Produce requests will be failed before the number of
retries has been exhausted if the timeout configured by delivery.timeout.ms
expires first before successful acknowledgement. Users should generally prefer
to leave this config unset and i [...]
+ "sourceType":
"org.apache.camel.component.kafka.springboot.KafkaComponentConfiguration"
},
{
"name": "camel.component.kafka.retry-backoff-ms",
diff --git
a/components-starter/camel-kafka-starter/src/main/java/org/apache/camel/component/kafka/springboot/KafkaComponentConfiguration.java
b/components-starter/camel-kafka-starter/src/main/java/org/apache/camel/component/kafka/springboot/KafkaComponentConfiguration.java
index f69ceb3d0c4..6b4ac7125f9 100644
---
a/components-starter/camel-kafka-starter/src/main/java/org/apache/camel/component/kafka/springboot/KafkaComponentConfiguration.java
+++
b/components-starter/camel-kafka-starter/src/main/java/org/apache/camel/component/kafka/springboot/KafkaComponentConfiguration.java
@@ -409,13 +409,19 @@ public class KafkaComponentConfiguration
*/
private Integer deliveryTimeoutMs = 120000;
/**
- * If set to 'true' the producer will ensure that exactly one copy of each
- * message is written in the stream. If 'false', producer retries may write
- * duplicates of the retried message in the stream. If set to true this
- * option will require max.in.flight.requests.per.connection to be set to 1
- * and retries cannot be zero and additionally acks must be set to 'all'.
- */
- private Boolean enableIdempotence = false;
+ * When set to 'true', the producer will ensure that exactly one copy of
+ * each message is written in the stream. If 'false', producer retries due
+ * to broker failures, etc., may write duplicates of the retried message in
+ * the stream. Note that enabling idempotence requires
+ * max.in.flight.requests.per.connection to be less than or equal to 5
(with
+ * message ordering preserved for any allowable value), retries to be
+ * greater than 0, and acks must be 'all'. Idempotence is enabled by
default
+ * if no conflicting configurations are set. If conflicting configurations
+ * are set and idempotence is not explicitly enabled, idempotence is
+ * disabled. If idempotence is explicitly enabled and conflicting
+ * configurations are set, a ConfigException is thrown.
+ */
+ private Boolean enableIdempotence = true;
/**
* To use a custom KafkaHeaderSerializer to serialize kafka headers values.
* The option is a
@@ -562,7 +568,7 @@ public class KafkaComponentConfiguration
/**
* The number of acknowledgments the producer requires the leader to have
* received before considering a request complete. This controls the
- * durability of records that are sent. The following settings are common:
+ * durability of records that are sent. The following settings are allowed:
* acks=0 If set to zero then the producer will not wait for any
* acknowledgment from the server at all. The record will be immediately
* added to the socket buffer and considered sent. No guarantee can be made
@@ -576,9 +582,12 @@ public class KafkaComponentConfiguration
* the record will be lost. acks=all This means the leader will wait for
the
* full set of in-sync replicas to acknowledge the record. This guarantees
* that the record will not be lost as long as at least one in-sync replica
- * remains alive. This is the strongest available guarantee.
+ * remains alive. This is the strongest available guarantee. This is
+ * equivalent to the acks=-1 setting. Note that enabling idempotence
+ * requires this config value to be 'all'. If conflicting configurations
are
+ * set and idempotence is not explicitly enabled, idempotence is disabled.
*/
- private String requestRequiredAcks = "1";
+ private String requestRequiredAcks = "all";
/**
* The amount of time the broker will wait trying to meet the
* request.required.acks requirement before sending back an error to the
@@ -589,12 +598,21 @@ public class KafkaComponentConfiguration
* Setting a value greater than zero will cause the client to resend any
* record whose send fails with a potentially transient error. Note that
* this retry is no different than if the client resent the record upon
- * receiving the error. Allowing retries will potentially change the
- * ordering of records because if two records are sent to a single
- * partition, and the first fails and is retried but the second succeeds,
- * then the second record may appear first.
- */
- private Integer retries = 0;
+ * receiving the error. Produce requests will be failed before the number
of
+ * retries has been exhausted if the timeout configured by
+ * delivery.timeout.ms expires first before successful acknowledgement.
+ * Users should generally prefer to leave this config unset and instead use
+ * delivery.timeout.ms to control retry behavior. Enabling idempotence
+ * requires this config value to be greater than 0. If conflicting
+ * configurations are set and idempotence is not explicitly enabled,
+ * idempotence is disabled. Allowing retries while setting
+ * enable.idempotence to false and max.in.flight.requests.per.connection to
+ * 1 will potentially change the ordering of records because if two batches
+ * are sent to a single partition, and the first fails and is retried but
+ * the second succeeds, then the records in the second batch may appear
+ * first.
+ */
+ private Integer retries;
/**
* Before each retry, the producer refreshes the metadata of relevant
topics
* to see if a new leader has been elected. Since leader election takes a
