|
|
|
|
@@ -39,7 +39,7 @@ To use Apache Kafka binder, you need to add `spring-cloud-stream-binder-kafka` a
|
|
|
|
|
</dependency>
|
|
|
|
|
----
|
|
|
|
|
|
|
|
|
|
Alternatively, you can also use the Spring Cloud Stream Kafka Starter, as shown inn the following example for Maven:
|
|
|
|
|
Alternatively, you can also use the Spring Cloud Stream Kafka Starter, as shown in the following example for Maven:
|
|
|
|
|
|
|
|
|
|
[source,xml]
|
|
|
|
|
----
|
|
|
|
|
@@ -60,7 +60,7 @@ The Apache Kafka Binder implementation maps each destination to an Apache Kafka
|
|
|
|
|
The consumer group maps directly to the same Apache Kafka concept.
|
|
|
|
|
Partitioning also maps directly to Apache Kafka partitions as well.
|
|
|
|
|
|
|
|
|
|
The binder currently uses the Apache Kafka `kafka-clients` 1.0.0 jar and is designed to be used with a broker of at least that version.
|
|
|
|
|
The binder currently uses the Apache Kafka `kafka-clients` version `2.3.1`.
|
|
|
|
|
This client can communicate with older brokers (see the Kafka documentation), but certain features may not be available.
|
|
|
|
|
For example, with versions earlier than 0.11.x.x, native headers are not supported.
|
|
|
|
|
Also, 0.11.x.x does not support the `autoAddPartitions` property.
|
|
|
|
|
@@ -155,21 +155,15 @@ Default: See individual producer properties.
|
|
|
|
|
|
|
|
|
|
spring.cloud.stream.kafka.binder.headerMapperBeanName::
|
|
|
|
|
The bean name of a `KafkaHeaderMapper` used for mapping `spring-messaging` headers to and from Kafka headers.
|
|
|
|
|
Use this, for example, if you wish to customize the trusted packages in a `DefaultKafkaHeaderMapper` that uses JSON deserialization for the headers.
|
|
|
|
|
Use this, for example, if you wish to customize the trusted packages in a `BinderHeaderMapper` bean that uses JSON deserialization for the headers.
|
|
|
|
|
If this custom `BinderHeaderMapper` bean is not made available to the binder using this property, then the binder will look for a header mapper bean with the name `kafkaBinderHeaderMapper` that is of type `BinderHeaderMapper` before falling back to a default `BinderHeaderMapper` created by the binder.
|
|
|
|
|
+
|
|
|
|
|
Default: none.
|
|
|
|
|
|
|
|
|
|
spring.cloud.stream.kafka.binder.authorizationExceptionRetryInterval::
|
|
|
|
|
Enables retrying in case of authorization exceptions.
|
|
|
|
|
Defines interval between each retry.
|
|
|
|
|
Accepts `Duration`, e.g. `30s`, `2m`, etc.
|
|
|
|
|
+
|
|
|
|
|
Default: `null` (retries disabled, fail fast)
|
|
|
|
|
|
|
|
|
|
[[kafka-consumer-properties]]
|
|
|
|
|
==== Kafka Consumer Properties
|
|
|
|
|
|
|
|
|
|
NOTE: To avoid repetition, Spring Cloud Stream supports setting values for all channels, in the format of `spring.cloud.stream.default.<property>=<value>`.
|
|
|
|
|
NOTE: To avoid repetition, Spring Cloud Stream supports setting values for all channels, in the format of `spring.cloud.stream.kafka.default.consumer.<property>=<value>`.
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
The following properties are available for Kafka consumers only and
|
|
|
|
|
@@ -234,9 +228,20 @@ The DLQ topic name can be configurable by setting the `dlqName` property.
|
|
|
|
|
This provides an alternative option to the more common Kafka replay scenario for the case when the number of errors is relatively small and replaying the entire original topic may be too cumbersome.
|
|
|
|
|
See <<kafka-dlq-processing>> processing for more information.
|
|
|
|
|
Starting with version 2.0, messages sent to the DLQ topic are enhanced with the following headers: `x-original-topic`, `x-exception-message`, and `x-exception-stacktrace` as `byte[]`.
|
|
|
|
|
By default, a failed record is sent to the same partition number in the DLQ topic as the original record.
|
|
|
|
|
See <<dlq-partition-selection>> for how to change that behavior.
|
|
|
|
|
**Not allowed when `destinationIsPattern` is `true`.**
|
|
|
|
|
+
|
|
|
|
|
Default: `false`.
|
|
|
|
|
dlqPartitions::
|
|
|
|
|
When `enableDlq` is true, and this property is not set, a dead letter topic with the same number of partitions as the primary topic(s) is created.
|
|
|
|
|
Usually, dead-letter records are sent to the same partition in the dead-letter topic as the original record.
|
|
|
|
|
This behavior can be changed; see <<dlq-partition-selection>>.
|
|
|
|
|
If this property is set to `1` and there is no `DqlPartitionFunction` bean, all dead-letter records will be written to partition `0`.
|
|
|
|
|
If this property is greater than `1`, you **MUST** provide a `DlqPartitionFunction` bean.
|
|
|
|
|
Note that the actual partition count is affected by the binder's `minPartitionCount` property.
|
|
|
|
|
+
|
|
|
|
|
Default: `none`
|
|
|
|
|
configuration::
|
|
|
|
|
Map with a key/value pair containing generic Kafka consumer properties.
|
|
|
|
|
In addition to having Kafka consumer properties, other configuration properties can be passed here.
|
|
|
|
|
@@ -311,7 +316,7 @@ Refer to the https://docs.spring.io/spring-kafka/docs/2.3.0.BUILD-SNAPSHOT/refer
|
|
|
|
|
[[kafka-producer-properties]]
|
|
|
|
|
==== Kafka Producer Properties
|
|
|
|
|
|
|
|
|
|
NOTE: To avoid repetition, Spring Cloud Stream supports setting values for all channels, in the format of `spring.cloud.stream.default.<property>=<value>`.
|
|
|
|
|
NOTE: To avoid repetition, Spring Cloud Stream supports setting values for all channels, in the format of `spring.cloud.stream.kafka.default.producer.<property>=<value>`.
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
The following properties are available for Kafka producers only and
|
|
|
|
|
@@ -576,6 +581,10 @@ When used in a processor application, the consumer starts the transaction; any r
|
|
|
|
|
When the listener exits normally, the listener container will send the offset to the transaction and commit it.
|
|
|
|
|
A common producer factory is used for all producer bindings configured using `spring.cloud.stream.kafka.binder.transaction.producer.*` properties; individual binding Kafka producer properties are ignored.
|
|
|
|
|
|
|
|
|
|
IMPORTANT: Normal binder retries (and dead lettering) are not supported with transactions because the retries will run in the original transaction, which may be rolled back and any published records will be rolled back too.
|
|
|
|
|
When retries are enabled (the common property `maxAttempts` is greater than zero) the retry properties are used to configure a `DefaultAfterRollbackProcessor` to enable retries at the container level.
|
|
|
|
|
Similarly, instead of publishing dead-letter records within the transaction, this functionality is moved to the listener container, again via the `DefaultAfterRollbackProcessor` which runs after the main transaction has rolled back.
|
|
|
|
|
|
|
|
|
|
If you wish to use transactions in a source application, or from some arbitrary thread for producer-only transaction (e.g. `@Scheduled` method), you must get a reference to the transactional producer factory and define a `KafkaTransactionManager` bean using it.
|
|
|
|
|
|
|
|
|
|
====
|
|
|
|
|
|
