Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Editing pass for spring-kafka

Browse Source 
I edited for clarity, punctuation, grammar, spelling, usage, and corporate voice.
I also added a few links (mostly to Javadoc).

Further changes to account for changes made while I was working

Someone made changes to a few paragraphs, so I edited the changed material, too.

* Fix "One Sentence per Line" violations
This commit is contained in:
Jay Bryant
2019-01-31 10:24:54 -06:00
committed by Artem Bilan
parent 18836db8b5
commit 84a7a706ab
10 changed files with 1265 additions and 826 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/reference/asciidoc/appendix.adoc
View File

@@ -1,8 +1,9 @@
[[deps-for-21x]]
== Override Dependencies to use the 2.1.x kafka-clients with an Embedded Broker
When using `spring-kafka-test` (version 2.2.x) with the 2.1.x `kafka-clients` jar, you will need to override certain transitive dependencies as follows:
When you use `spring-kafka-test` (version 2.2.x) with the 2.1.x `kafka-clients` jar, you need to override certain transitive dependencies, as follows:
====
[source, xml]
----
<dependency>
@@ -46,6 +47,7 @@ When using `spring-kafka-test` (version 2.2.x) with the 2.1.x `kafka-clients` ja
<scope>test</scope>
</dependency>
----
====
[appendix]
[[history]]

78
src/reference/asciidoc/changes-since-1.0.adoc
View File

@@ -5,59 +5,65 @@
This version requires the 1.0.0 `kafka-clients` or higher.
NOTE: The 1.1.x client is supported, with _version 2.1.5_, but you will need to override dependencies as described in <<deps-for-11x>>.
The 1.1.x client will be supported natively in _version 2.2_.
NOTE: The 1.1.x client is supported with version 2.1.5, but you need to override dependencies as described in <<deps-for-11x>>.
// TODO: No topic marked deps-for-11x exists in any of the files in the directory that contains this file.
The 1.1.x client is supported natively in version 2.2.
==== JSON Improvements
The `StringJsonMessageConverter` and `JsonSerializer` now add type information in `Headers`, allowing the converter and `JsonDeserializer` to create specific types on reception, based on the message itself rather than a fixed configured type.
The `StringJsonMessageConverter` and `JsonSerializer` now add type information in `Headers`, letting the converter and `JsonDeserializer` create specific types on reception, based on the message itself rather than a fixed configured type.
See <<serdes>> for more information.
==== Container Stopping Error Handlers
Container Error handlers are now provided for both record and batch listeners that treat any exceptions thrown by the listener as fatal; they stop the container.
Container error handlers are now provided for both record and batch listeners that treat any exceptions thrown by the listener as fatal/
They stop the container.
See <<annotation-error-handling>> for more information.
==== Pausing/Resuming Containers
==== Pausing and Resuming Containers
The listener containers now have `pause()` and `resume()` methods (since _version 2.1.3_).
The listener containers now have `pause()` and `resume()` methods (since version 2.1.3).
See <<pause-resume>> for more information.
==== Stateful Retry
Starting with _version 2.1.3_, stateful retry can be configured; see <<stateful-retry>> for more information.
Starting with version 2.1.3, you can configure stateful retry.
See <<stateful-retry>> for more information.
==== Client ID
Starting with _version 2.1.1_, it is now possible to set the `client.id` prefix on `@KafkaListener`.
Previously, to customize the client id, you would need a separate consumer factory (and container factory) per listener.
The prefix is suffixed with `-n` to provide unique client ids when using concurrency.
Starting with version 2.1.1, you can now set the `client.id` prefix on `@KafkaListener`.
Previously, to customize the client ID, you needed a separate consumer factory (and container factory) per listener.
The prefix is suffixed with `-n` to provide unique client IDs when you use concurrency.
==== Logging Offset Commits
By default, logging of topic offset commits is performed with the DEBUG logging level.
Starting with _version 2.1.2_, there is a new property in `ContainerProperties` called `commitLogLevel` which allows you to specify the log level for these messages.
By default, logging of topic offset commits is performed with the `DEBUG` logging level.
Starting with version 2.1.2, a new property in `ContainerProperties` called `commitLogLevel` lets you specify the log level for these messages.
See <<kafka-container>> for more information.
==== Default @KafkaHandler
Starting with _version 2.1.3_, one of the `@KafkaHandler` s on a class-level `@KafkaListener` can be designated as the default.
Starting with version 2.1.3, you can designate one of the `@KafkaHandler` annotations on a class-level `@KafkaListener` as the default.
See <<class-level-kafkalistener>> for more information.
==== ReplyingKafkaTemplate
Starting with _version 2.1.3_, a subclass of `KafkaTemplate` is provided to support request/reply semantics.
Starting with version 2.1.3, a subclass of `KafkaTemplate` is provided to support request/reply semantics.
See <<replying-template>> for more information.
==== ChainedKafkaTransactionManager
_version 2.1.3_ introduced the `ChainedKafkaTransactionManager` see <<chained-transaction-manager>> for more information.
Version 2.1.3 introduced the `ChainedKafkaTransactionManager`.
See <<chained-transaction-manager>> for more information.
==== Migration Guide from 2.0
https://github.com/spring-projects/spring-kafka/wiki/Spring-for-Apache-Kafka-2.0-to-2.1-Migration-Guide[2.0 to 2.1 Migration].
See the https://github.com/spring-projects/spring-kafka/wiki/Spring-for-Apache-Kafka-2.0-to-2.1-Migration-Guide[2.0 to 2.1 Migration] guide.
=== Changes Between 1.3 and 2.0
@@ -65,9 +71,9 @@ https://github.com/spring-projects/spring-kafka/wiki/Spring-for-Apache-Kafka-2.0
The Spring for Apache Kafka project now requires Spring Framework 5.0 and Java 8.
==== @KafkaListener Changes
==== `@KafkaListener` Changes
You can now annotate `@KafkaListener` methods (and classes, and `@KafkaHandler` methods) with `@SendTo`.
You can now annotate `@KafkaListener` methods (and classes and `@KafkaHandler` methods) with `@SendTo`.
If the method returns a result, it is forwarded to the specified topic.
See <<annotation-send-to>> for more information.
@@ -76,7 +82,7 @@ See <<annotation-send-to>> for more information.
Message listeners can now be aware of the `Consumer` object.
See <<message-listeners>> for more information.
==== ConsumerAwareRebalanceListener
==== Using `ConsumerAwareRebalanceListener`
Rebalance listeners can now access the `Consumer` object during rebalance notifications.
See <<rebalance-listeners>> for more information.
@@ -85,40 +91,42 @@ See <<rebalance-listeners>> for more information.
==== Support for Transactions
The 0.11.0.0 client library added support for transactions; the `KafkaTransactionManager` and other support for transactions has been added.
The 0.11.0.0 client library added support for transactions.
The `KafkaTransactionManager` and other support for transactions have been added.
See <<transactions>> for more information.
==== Support for Headers
The 0.11.0.0 client library added support for message headers; these can now be mapped to/from `spring-messaging` `MessageHeaders`.
The 0.11.0.0 client library added support for message headers.
These can now be mapped to and from `spring-messaging` `MessageHeaders`.
See <<headers>> for more information.
==== Creating Topics
The 0.11.0.0 client library provides an `AdminClient` which can be used to create topics.
The `KafkaAdmin` uses this client to automatically add topics defined as `@Bean` s.
The 0.11.0.0 client library provides an `AdminClient`, which you can use to create topics.
The `KafkaAdmin` uses this client to automatically add topics defined as `@Bean` instances.
==== Support for Kafka timestamps
==== Support for Kafka Timestamps
`KafkaTemplate` now supports API to add records with timestamps.
`KafkaTemplate` now supports an API to add records with timestamps.
New `KafkaHeaders` have been introduced regarding `timestamp` support.
Also new `KafkaConditions.timestamp()` and `KafkaMatchers.hasTimestamp()` testing utilities have been added.
See <<kafka-template>>, <<kafka-listener-annotation>> and <<testing>> for more details.
Also, new `KafkaConditions.timestamp()` and `KafkaMatchers.hasTimestamp()` testing utilities have been added.
See <<kafka-template>>, <<kafka-listener-annotation>>, and <<testing>> for more details.
==== @KafkaListener Changes
==== `@KafkaListener` Changes
You can now configure a `KafkaListenerErrorHandler` to handle exceptions.
See <<annotation-error-handling>> for more information.
By default, the `@KafkaListener` `id` property is now used as the `group.id` property, overriding the property configured in the consumer factory (if present).
Further, you can explicitly configure the `groupId` on the annotation.
Previously, you would have needed a separate container factory (and consumer factory) to use different `group.id` s for listeners.
Previously, you would have needed a separate container factory (and consumer factory) to use different `group.id` values for listeners.
To restore the previous behavior of using the factory configured `group.id`, set the `idIsGroup` property on the annotation to `false`.
==== @EmbeddedKafka Annotation
==== `@EmbeddedKafka` Annotation
For convenience a test class level `@EmbeddedKafka` annotation is provided with the purpose to register `KafkaEmbedded` as a bean.
For convenience, a test class-level `@EmbeddedKafka` annotation is provided, to register `KafkaEmbedded` as a bean.
See <<testing>> for more information.
==== Kerberos Configuration
@@ -143,7 +151,7 @@ Listeners can be configured to receive the entire batch of messages returned by
==== Null Payloads
Null payloads are used to "delete" keys when using log compaction.
Null payloads are used to "`delete`" keys when you use log compaction.
==== Initial Offset
@@ -151,7 +159,7 @@ When explicitly assigning partitions, you can now configure the initial offset r
==== Seek
You can now seek the position of each topic/partition.
This can be used to set the initial position during initialization when group management is in use and Kafka assigns the partitions.
You can also seek when an idle container is detected, or at any arbitrary point in your application's execution.
You can now seek the position of each topic or partition.
You can use this to set the initial position during initialization when group management is in use and Kafka assigns the partitions.
You can also seek when an idle container is detected or at any arbitrary point in your application's execution.
See <<seek>> for more information.

10
src/reference/asciidoc/index.adoc
View File

@@ -1,7 +1,5 @@
[[spring-kafka-reference]]
= Spring for Apache Kafka
:toc:
== Preface
@@ -18,12 +16,10 @@ include::./whats-new.adoc[]
== Introduction
This first part of the reference documentation is a high-level overview of Spring for Apache Kafka and the underlying
concepts and some code snippets that will get you up and running as quickly as possible.
concepts and some code snippets that can help you get up and running as quickly as possible.
include::quick-tour.adoc[]
// include::whats-new.adoc[]
== Reference
This part of the reference documentation details the various components that comprise Spring for Apache Kafka.
@@ -37,7 +33,7 @@ include::testing.adoc[]
== Spring Integration
This part of the reference shows how to use the `spring-integration-kafka` module of Spring Integration.
This part of the reference guide shows how to use the `spring-integration-kafka` module of Spring Integration.
include::si-kafka.adoc[]
@@ -45,7 +41,7 @@ include::si-kafka.adoc[]
== Other Resources
In addition to this reference documentation, there exist a number of other resources that may help you learn about
In addition to this reference documentation, we recommend a number of other resources that may help you learn about
Spring and Apache Kafka.
- https://kafka.apache.org/[Apache Kafka Project Home Page]

1458
src/reference/asciidoc/kafka.adoc
View File

File diff suppressed because it is too large Load Diff

2
src/reference/asciidoc/preface.adoc
View File

@@ -1,4 +1,4 @@
[[preface]]
The Spring for Apache Kafka project applies core Spring concepts to the development of Kafka-based messaging solutions.
We provide a "template" as a high-level abstraction for sending messages.
We provide a "`template`" as a high-level abstraction for sending messages.
We also provide support for Message-driven POJOs.

50
src/reference/asciidoc/quick-tour.adoc
View File

@@ -1,14 +1,14 @@
[[quick-tour]]
=== Quick Tour for the Impatient
==== Introduction
This is the five-minute tour to get started with Spring Kafka.
This is the 5 minute tour to get started with Spring Kafka.
Prerequisites: install and run Apache Kafka
Then grab the spring-kafka JAR and all of its dependencies - the easiest way to do that is to declare a dependency in
your build tool, e.g. for Maven:
Prerequisites: You must install and run Apache Kafka.
Then you must grab the spring-kafka JAR and all of its dependencies.
The easiest way to do that is to declare a dependency in your build tool.
The following example shows how to do so with Maven:
====
[source,xml,subs="+attributes"]
----
<dependency>
@@ -17,25 +17,31 @@ your build tool, e.g. for Maven:
<version>{spring-kafka-version}</version>
</dependency>
----
====
And for Gradle:
The following example shows how to do so with Gradle:
====
[source,groovy,subs="+attributes"]
----
compile 'org.springframework.kafka:spring-kafka:{spring-kafka-version}'
----
====
[[compatibility]]
===== Compatibility
- Apache Kafka Clients 2.0.0
- Spring Framework 5.1.x
- Minimum Java version: 8
This quick tour works with the following versions:
===== Very, Very Quick
* Apache Kafka Clients 2.0.0
* Spring Framework 5.1.x
* Minimum Java version: 8
Using plain Java to send and receive a message:
===== A Very, Very Quick Example
As the following example shows, you can use plain Java to send and receive a message:
====
[source,java]
----
@Test
@@ -114,11 +120,14 @@ private Map<String, Object> senderProps() {
return props;
}
----
====
===== With Java Configuration
A similar example but with Spring configuration in Java:
You can do the same work as appears in the previous example with Spring configuration in Java.
The following example shows how to do so:
====
[source,java]
----
@Autowired
@@ -199,12 +208,14 @@ public class Listener {
}
----
====
===== Even Quicker, with Spring Boot
The following Spring Boot application sends 3 messages to a topic, receives them, and stops.
Spring Boot can make things even simpler.
The following Spring Boot application sends three messages to a topic, receives them, and stops:
.Application
====
[source, java]
----
@SpringBootApplication
@@ -238,14 +249,19 @@ public class Application implements CommandLineRunner {
}
----
====
Boot takes care of most of the configuration; when using a local broker, the only properties we need are:
Boot takes care of most of the configuration.
When we use a local broker, the only properties we need are the following:
.application.properties
====
[source]
----
spring.kafka.consumer.group-id=foo
spring.kafka.consumer.auto-offset-reset=earliest
----
====
The first because we are using group management to assign topic partitions to consumers so we need a group, the second to ensure the new consumer group will get the messages we just sent, because the container might start after the sends have completed.
We need the first property because we are using group management to assign topic partitions to consumers, so we need a group.
The second property ensures the new consumer group gets the messages we sent, because the container might start after the sends have completed.

201
src/reference/asciidoc/si-kafka.adoc
View File

@@ -1,57 +1,63 @@
[[si-kafka]]
=== Spring Integration for Apache Kafka
==== Introduction
This documentation pertains to versions 2.0.0 and above; for documentation for earlier releases, see the https://github.com/spring-projects/spring-integration-kafka/blob/1.3.x/README.md[1.3.x README].
This documentation pertains to versions 2.0.0 and above.
For documentation for earlier releases, see the https://github.com/spring-projects/spring-integration-kafka/blob/1.3.x/README.md[1.3.x README].
Spring Integration Kafka is now based on the http://projects.spring.io/spring-kafka/[Spring for Apache Kafka project].
It provides the following components:
- Outbound Channel Adapter
- Message-Driven Channel Adapter
These are discussed in the following sections.
* <<si-outbound>>
* <<si-inbound>>
[[si-outbound]]
==== Outbound Channel Adapter
The Outbound channel adapter is used to publish messages from a Spring Integration channel to Kafka topics.
The channel is defined in the application context and then wired into the application that sends messages to Kafka.
Sender applications can publish to Kafka via Spring Integration messages, which are internally converted
to Kafka messages by the outbound channel adapter, as follows: the payload of the Spring Integration message will be
used to populate the payload of the Kafka message, and (by default) the `kafka_messageKey` header of the Spring
Integration message will be used to populate the key of the Kafka message.
Sender applications can publish to Kafka by using Spring Integration messages, which are internally converted to Kafka messages by the outbound channel adapter, as follows:
The target topic and partition for publishing the message can be customized through the `kafka_topic`
* The payload of the Spring Integration message is used to populate the payload of the Kafka message.
* By default, the `kafka_messageKey` header of the Spring Integration message is used to populate the key of the Kafka message.
You can customize the target topic and partition for publishing the message through the `kafka_topic`
and `kafka_partitionId` headers, respectively.
In addition, the `<int-kafka:outbound-channel-adapter>` provides the ability to extract the key, target topic, and
target partition by applying SpEL expressions on the outbound message. To that end, it supports the mutually exclusive
pairs of attributes `topic`/`topic-expression`, `message-key`/`message-key-expression`, and
`partition-id`/`partition-id-expression`, to allow the specification of `topic`,`message-key` and `partition-id`
respectively as static values on the adapter, or to dynamically evaluate their values at runtime against
the request message.
In addition, the `<int-kafka:outbound-channel-adapter>` provides the ability to extract the key, target topic, and target partition by applying SpEL expressions on the outbound message.
To that end, it supports three mutually exclusive pairs of attributes:
* `topic` and `topic-expression`
* `message-key` and `message-key-expression`
* `partition-id` and `partition-id-expression`
These let you specify `topic`, `message-key`, and `partition-id`, respectively, as static values on the adapter or to dynamically evaluate their values at runtime against the request message.
IMPORTANT: The `KafkaHeaders` interface (provided by `spring-kafka`) contains constants used for interacting with
headers.
The `messageKey` and `topic` default headers now require a `kafka_` prefix.
When migrating from an earlier version that used the old headers, you need to specify
`message-key-expression="headers['messageKey']"` and `topic-expression="headers['topic']"` on the
`<int-kafka:outbound-channel-adapter>`, or simply change the headers upstream to
the new headers from `KafkaHeaders` using a `<header-enricher>` or `MessageBuilder`.
Or, of course, configure them on the adapter using `topic` and `message-key` if you are using constant values.
`<int-kafka:outbound-channel-adapter>`.
Alternatively, you can change the headers upstream to
the new headers from `KafkaHeaders` by using a `<header-enricher>` or a `MessageBuilder`.
If you use constant values, you can also configure them on the adapter by using `topic` and `message-key`.
NOTE : If the adapter is configured with a topic or message key (either with a constant or expression), those are used
and the corresponding header is ignored.
If you wish the header to override the configuration, you need to configure it in an expression, such as:
If you wish the header to override the configuration, you need to configure it in an expression, such as the following:
`topic-expression="headers['topic'] != null ? headers['topic'] : 'myTopic'"`.
====
[source,java]
----
topic-expression="headers['topic'] != null ? headers['topic'] : 'myTopic'"
----
====
The adapter requires a `KafkaTemplate`.
Here is an example of how the Kafka outbound channel adapter is configured with XML:
The following example shows how to configure the Kafka outbound channel adapter with XML:
====
[source, xml]
----
<int-kafka:outbound-channel-adapter id="kafkaOutboundChannelAdapter"
@@ -80,11 +86,13 @@ Here is an example of how the Kafka outbound channel adapter is configured with
</constructor-arg>
</bean>
----
====
As you can see, the adapter requires a `KafkaTemplate` which, in turn, requires a suitably configured `KafkaProducerFactory`.
The adapter requires a `KafkaTemplate`, which, in turn, requires a suitably configured `KafkaProducerFactory`.
When using Java Configuration:
The following example shows how to configure the Kafka outbound channel adapter with Java:
====
[source, java]
----
@Bean
@@ -112,9 +120,11 @@ public ProducerFactory<String, String> producerFactory() {
return new DefaultKafkaProducerFactory<>(props);
}
----
====
When using Spring Integration Java DSL:
The following example shows how to configure the Kafka outbound channel adapter Spring Integration Java DSL:
====
[source, java]
----
@Bean
@@ -156,24 +166,28 @@ private KafkaProducerMessageHandlerSpec<Integer, String, ?> kafkaMessageHandler(
.configureKafkaTemplate(t -> t.id("kafkaTemplate:" + topic));
}
----
====
If a `send-failure-channel` (`sendFailureChannel`) is provided and a send failure is received (sync or async), an `ErrorMessage` is sent to the channel.
The payload is a `KafkaSendFailureException` with properties `failedMessage`, `record` (the `ProducerRecord`) and `cause`.
The `DefaultErrorMessageStrategy` can be overridden via the `error-message-strategy` property.
If a `send-failure-channel` (`sendFailureChannel`) is provided and a send failure (sync or async) is received, an `ErrorMessage` is sent to the channel.
The payload is a `KafkaSendFailureException` with `failedMessage`, `record` (the `ProducerRecord`) and `cause` properties.
You can override the `DefaultErrorMessageStrategy` by setting the `error-message-strategy` property.
If a `send-success-channel` (`sendSuccessChannel`) is provided, a message with a payload of type `org.apache.kafka.clients.producer.RecordMetadata` will be sent after a successful send.
If a `send-success-channel` (`sendSuccessChannel`) is provided, a message with a payload of type `org.apache.kafka.clients.producer.RecordMetadata` is sent after a successful send.
[[si-inbound]]
==== Message Driven Channel Adapter
==== Message-driven Channel Adapter
The `KafkaMessageDrivenChannelAdapter` (`<int-kafka:message-driven-channel-adapter>`) uses a `spring-kafka` `KafkaMessageListenerContainer` or `ConcurrentListenerContainer`.
Starting with _spring-integration-kafka version 2.1_, the `mode` attribute is available (`record` or `batch`, default `record`).
For `record` mode, each message payload is converted from a single `ConsumerRecord`; for mode `batch` the payload is a list of objects which are converted from all the `ConsumerRecord` s returned by the consumer poll.
As with the batched `@KafkaListener`, the `KafkaHeaders.RECEIVED_MESSAGE_KEY`, `KafkaHeaders.RECEIVED_PARTITION_ID`, `KafkaHeaders.RECEIVED_TOPIC` and `KafkaHeaders.OFFSET` headers are also lists with, positions corresponding to the position in the payload.
Starting with spring-integration-kafka version 2.1, the `mode` attribute is available.
It can accept values of `record` or `batch` (default: `record`).
For `record` mode, each message payload is converted from a single `ConsumerRecord`.
For `batch` mode, the payload is a list of objects that are converted from all the `ConsumerRecord` instances returned by the consumer poll.
As with the batched `@KafkaListener`, the `KafkaHeaders.RECEIVED_MESSAGE_KEY`, `KafkaHeaders.RECEIVED_PARTITION_ID`, `KafkaHeaders.RECEIVED_TOPIC`, and `KafkaHeaders.OFFSET` headers are also lists, with positions corresponding to the position in the payload.
An example of xml configuration variant is shown here:
The following example shows how to configure a message-driven channel adapter with XML:
====
[source, xml]
----
<int-kafka:message-driven-channel-adapter
@@ -208,9 +222,11 @@ An example of xml configuration variant is shown here:
</bean>
----
====
When using Java Configuration:
The following example shows how to configure a message-driven channel adapter with Java:
====
[source, java]
----
@Bean
@@ -237,9 +253,11 @@ public ConsumerFactory<String, String> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(props);
}
----
====
When using Spring Integration Java DSL:
The following example shows how to configure a message-driven channel adapter with the Spring Integration Java DSL:
====
[source, java]
----
@Bean
@@ -262,25 +280,30 @@ public IntegrationFlow topic1ListenerFromKafkaFlow() {
.get();
}
----
====
Received messages will have certain headers populated.
Refer to the `KafkaHeaders` class for more information.
Received messages have certain headers populated.
See the https://docs.spring.io/spring-kafka/api/org/springframework/kafka/support/KafkaHeaders.html[`KafkaHeaders` class] for more information.
IMPORTANT: The `Consumer` object (in the `kafka_consumer` header) is not thread-safe; you must only invoke its methods on the thread that calls the listener within the adapter; if you hand off the message to another thread, you must not call its methods.
IMPORTANT: The `Consumer` object (in the `kafka_consumer` header) is not thread-safe.
You must invoke its methods only on the thread that calls the listener within the adapter.
If you hand off the message to another thread, you must not call its methods.
When a `retry-template` is provided, delivery failures will be retried according to its retry policy.
When a `retry-template` is provided, delivery failures are retried according to its retry policy.
An `error-channel` is not allowed in this case.
The `recovery-callback` can be used to handle the error when retries are exhausted.
In most cases, this will be an `ErrorMessageSendingRecoverer` which will send the `ErrorMessage` to a channel.
You can use the `recovery-callback` to handle the error when retries are exhausted.
In most cases, this is an `ErrorMessageSendingRecoverer` that sends the `ErrorMessage` to a channel.
When building `ErrorMessage` (for use in the `error-channel` or `recovery-callback`), you can customize the error message using the `error-message-strategy` property.
By default, a `RawRecordHeaderErrorMessageStrategy` is used; providing access to the converted message as well as the raw `ConsumerRecord`.
When building an `ErrorMessage` (for use in the `error-channel` or `recovery-callback`), you can customize the error message by setting the `error-message-strategy` property.
By default, a `RawRecordHeaderErrorMessageStrategy` is used, to provide access to the converted message as well as the raw `ConsumerRecord`.
Starting with _Spring for Apache Kafka version 2.2_ (_Spring Integration Kafka 3.1_), the container factory used for `@KafkaListener` annotations can also be used to create `ConcurrentMessageListenerContainer` s for other purposes.
Starting with Spring for Apache Kafka version 2.2 (Spring Integration Kafka 3.1), you can also use the container factory that is used for `@KafkaListener` annotations to create `ConcurrentMessageListenerContainer` instances for other purposes.
See <<container-factory>> for an example.
With the Java DSL, the container does not have to be configured as a `@Bean` because the DSL will register the container as a bean.
With the Java DSL, the container does not have to be configured as a `@Bean`, because the DSL registers the container as a bean.
The following example shows how to do so:
====
[source, java]
----
@Bean
@@ -293,21 +316,25 @@ public IntegrationFlow topic2ListenerFromKafkaFlow() {
get();
}
----
====
Notice that, in this case, the adapter is given an `id` ("topic2Adapter"); the container will be registered in the application context with the name `topic2Adapter.container`.
If the adapter does not have an `id` property, the container's bean name will be the container's fully qualified class name + `#n` where `n` is incremented for each container.
Notice that, in this case, the adapter is given an `id` (`topic2Adapter`).
The container is registered in the application context with a name of `topic2Adapter.container`.
If the adapter does not have an `id` property, the container's bean name is the container's fully qualified class name plus `#n`, where `n` is incremented for each container.
[[si-outbound-gateway]]
==== Outbound Gateway
The outbound gateway is for request/reply operations; it is different to most Spring Integration gateways in that the sending thread does not block in the gateway, the reply is processed on the reply listener container thread.
Of course, if user code invokes the gateway behind a synchronous https://docs.spring.io/spring-integration/reference/html/messaging-endpoints-chapter.html#gateway[Messaging Gateway], the user thread will block there until the reply is received (or a timeout occurs).
The outbound gateway is for request/reply operations.
It differs from most Spring Integration gateways in that the sending thread does not block in the gateway and the reply is processed on the reply listener container thread.
If your code invokes the gateway behind a synchronous https://docs.spring.io/spring-integration/reference/html/messaging-endpoints-chapter.html#gateway[Messaging Gateway], the user thread blocks there until the reply is received (or a timeout occurs).
IMPORTANT: the gateway will not accept requests until the reply container has been assigned its topics and partitions.
IMPORTANT: The gateway does not accept requests until the reply container has been assigned its topics and partitions.
It is suggested that you add a `ConsumerRebalanceListener` to the template's reply container properties and wait for the `onPartitionsAssigned` call before sending messages to the gateway.
Here is an example of configuring a gateway, with Java Configuration:
The following example shows how to configure a gateway with Java:
====
[source, java]
----
@Bean
@@ -317,20 +344,23 @@ public KafkaProducerMessageHandler<String, String> outGateway(
return new KafkaProducerMessageHandler<>(kafkaTemplate);
}
----
====
Notice that the same class as the <<si-outbound, outbound channel adapter>> is used, the only difference being that the kafka template passed into the constructor is a `ReplyingKafkaTemplate` - see <<replying-template>> for more information.
Notice that the same class as the <<si-outbound,outbound channel adapter>> is used, the only difference being that the Kafka template passed into the constructor is a `ReplyingKafkaTemplate`.
See <<replying-template>> for more information.
The outbound topic, partition, key etc, are determined the same way as the outbound adapter.
The outbound topic, partition, key, and so on are determined in the same way as the outbound adapter.
The reply topic is determined as follows:
1. A message header `KafkaHeaders.REPLY_TOPIC`, if present (must have a `String` or `byte[]` value) - validated against the template's reply container subscribed topics.
2. If the template's `replyContainer` is subscribed to just one topic, it will be used.
. A message header named `KafkaHeaders.REPLY_TOPIC` (if present, it must have a `String` or `byte[]` value) is validated against the template's reply container's subscribed topics.
. If the template's `replyContainer` is subscribed to only one topic, it is used.
You can also specify a `KafkaHeaders.REPLY_PARTITION` header to determine a specific partition to be used for replies.
Again, this is validated against the template's reply container subscriptions.
Again, this is validated against the template's reply container's subscriptions.
Configuring with the Java DSL:
The following example shows how to configure an outbound gateway with the Java DSL:
====
[source, java]
----
@Bean
@@ -342,9 +372,11 @@ public IntegrationFlow outboundGateFlow(
.get();
}
----
====
Or:
Alternatively, you can also use a configuration similar to the following bean:
====
[source, java]
----
@Bean
@@ -356,16 +388,18 @@ public IntegrationFlow outboundGateFlow() {
.get();
}
----
====
XML configuration is not currently available for this component.
NOTE: XML configuration is not currently available for this component.
[[si-inbound-gateway]]
==== Inbound Gateway
The inbound gateway is for request/reply operations.
Configuring an inbound gateway with Java Configuration:
The following example shows how to configure an inbound gateway with Java:
====
[source, java]
----
@Bean
@@ -381,9 +415,11 @@ public KafkaInboundGateway<Integer, String, String> inboundGateway(
return gateway;
}
----
====
Configuring a simple upper case converter with the Java DSL:
The following example shows how to configure a simple upper case converter with the Java DSL:
====
[source, java]
----
@Bean
@@ -397,9 +433,11 @@ public IntegrationFlow serverGateway(
.get();
}
----
====
Or:
Alternatively, you could configure an upper-case converter by using code similar to the following:
====
[source, java]
----
@Bean
@@ -412,20 +450,24 @@ public IntegrationFlow serverGateway() {
.get();
}
----
====
XML configuration is not currently available for this component.
NOTE: XML configuration is not currently available for this component.
Starting with _Spring for Apache Kafka version 2.2_ (_Spring Integration Kafka 3.1_), the container factory used for `@KafkaListener` annotations can also be used to create `ConcurrentMessageListenerContainer` s for other purposes.
Starting with Spring for Apache Kafka version 2.2 (Spring Integration Kafka 3.1), you can also use the container factory that is used for `@KafkaListener` annotations to create `ConcurrentMessageListenerContainer` instances for other purposes.
See <<container-factory>> and <<si-inbound>> for examples.
[[message-conversion]]
==== Message Conversion
A `StringJsonMessageConverter` is provided, see <<serdes>> for more information.
A `StringJsonMessageConverter` is provided.
See <<serdes>> for more information.
When using this converter with a message-driven channel adapter, you can specify the type to which you want the incoming payload to be converted.
This is achieved by setting the `payload-type` attribute (`payloadType` property) on the adapter.
The following example shows how to do so in XML configuration:
====
[source, xml]
----
<int-kafka:message-driven-channel-adapter
@@ -443,7 +485,11 @@ This is achieved by setting the `payload-type` attribute (`payloadType` property
class="org.springframework.kafka.support.converter.MessagingMessageConverter"/>
----
====
The following example shows how to set the `payload-type` attribute (`payloadType` property) on the adapter in Java configuration:
====
[source, java]
----
@Bean
@@ -457,15 +503,18 @@ public KafkaMessageDrivenChannelAdapter<String, String>
return kafkaMessageDrivenChannelAdapter;
}
----
====
[[si-tombstones]]
==== Null Payloads and Log Compaction 'Tombstone' Records
Spring Messaging `Message<?>` objects cannot have `null` payloads; when using the Kafka endpoints, `null` payloads (also known as tombstone records) are represented by a payload of type `KafkaNull`.
Spring Messaging `Message<?>` objects cannot have `null` payloads.
When you use the Kafka endpoints, `null` payloads (also known as tombstone records) are represented by a payload of type `KafkaNull`.
See <<tombstones>> for more information.
Starting with version 3.1 of Spring Integration Kafka, such records can now be received by Spring Integration POJO methods with a true `null` value instead.
Simply mark the parameter with `@Payload(required = false)`.
To do so, mark the parameter with `@Payload(required = false)`.
The following example shows how to do so:
====
[source, java]
@@ -487,11 +536,11 @@ See the http://projects.spring.io/spring-kafka/[Spring for Apache Kafka Project
The 2.1.x branch introduced the following changes:
* Update to `spring-kafka` 1.1.x; including support of batch payloads
* Support `sync` outbound requests via XML configuration
* Update to `spring-kafka` 1.1.x, including support of batch payloads
* Support `sync` outbound requests in XML configuration
* Support `payload-type` for inbound channel adapters
* Support for Enhanced Error handling for the inbound channel adapter (2.1.1)
* Support for send success/failure messages (2.1.2)
* Support for enhanced error handling for the inbound channel adapter (2.1.1)
* Support for send success and failure messages (2.1.2)
===== 2.2.x
@@ -503,7 +552,7 @@ The 2.2.x branch introduced the following changes:
The 2.3.x branch introduced the following changes:
* Update to `spring-kafka` 1.3.x; including support for transactions and header mapping provided by `kafka-clients` 0.11.0.0
* Update to `spring-kafka` 1.3.x, including support for transactions and header mapping provided by `kafka-clients` 0.11.0.0
* Support for record timestamps
===== 3.0.x
@@ -511,7 +560,7 @@ The 2.3.x branch introduced the following changes:
* Update to `spring-kafka` 2.1.x and `kafka-clients` 1.0.0
* Support `ConsumerAwareMessageListener` (`Consumer` is available in a message header)
* Update to Spring Integration 5.0 and Java 8
* Moved Java DSL to main project
* Moved Java DSL to the main project
* Added inbound and outbound gateways (3.0.2)
===== 3.1.x

115
src/reference/asciidoc/streams.adoc
View File

@@ -1,16 +1,15 @@
[[kafka-streams]]
=== Kafka Streams Support
==== Introduction
Starting with _version 1.1.4_, Spring for Apache Kafka provides first class support for https://kafka.apache.org/documentation/streams[Kafka Streams].
For using it from a Spring application, the `kafka-streams` jar must be present on classpath.
It is an optional dependency of the `spring-kafka` project and isn't downloaded transitively.
Starting with version 1.1.4, Spring for Apache Kafka provides first-class support for https://kafka.apache.org/documentation/streams[Kafka Streams].
To use it from a Spring application, the `kafka-streams` jar must be present on classpath.
It is an optional dependency of the `spring-kafka` project and is not downloaded transitively.
==== Basics
The reference Apache Kafka Streams documentation suggests this way of using the API:
The reference Apache Kafka Streams documentation suggests the following way of using the API:
====
[source, java]
----
// Use the builders to define the actual processing topology, e.g. to specify
@@ -32,17 +31,25 @@ streams.start();
// Stop the Kafka Streams instance
streams.close();
----
====
So, we have two main components: `StreamsBuilder` with an API to build `KStream` (or `KTable`) instances and `KafkaStreams` to manage their lifecycle.
Note: all `KStream` instances exposed to a `KafkaStreams` instance by a single `StreamsBuilder` will be started and stopped at the same time, even if they have a fully different logic.
In other words all our streams defined by a `StreamsBuilder` are tied with a single lifecycle control.
Once a `KafkaStreams` instance has been closed via `streams.close()` it cannot be restarted, and a new `KafkaStreams` instance to restart stream processing must be created instead.
So, we have two main components:
* `StreamsBuilder`: With an API to build `KStream` (or `KTable`) instances.
* `KafkaStreams`: To manage the lifecycle of those instances.
NOTE: All `KStream` instances exposed to a `KafkaStreams` instance by a single `StreamsBuilder` are started and stopped at the same time, even if they have different logic.
In other words, all streams defined by a `StreamsBuilder` are tied with a single lifecycle control.
Once a `KafkaStreams` instance has been closed by `streams.close()`, it cannot be restarted.
Instead, a new `KafkaStreams` instance to restart stream processing must be created.
==== Spring Management
To simplify the usage of Kafka Streams from the Spring application context perspective and utilize the lifecycle management via container, the Spring for Apache Kafka introduces `StreamsBuilderFactoryBean`.
This is an `AbstractFactoryBean` implementation to expose a `StreamsBuilder` singleton instance as a bean:
To simplify using Kafka Streams from the Spring application context perspective and use the lifecycle management through a container, the Spring for Apache Kafka introduces `StreamsBuilderFactoryBean`.
This is an `AbstractFactoryBean` implementation to expose a `StreamsBuilder` singleton instance as a bean.
The following example creates such a bean:
====
[source, java]
----
@Bean
@@ -50,14 +57,18 @@ public FactoryBean<StreamsBuilderFactoryBean> myKStreamBuilder(KafkaStreamsConfi
return new StreamsBuilderFactoryBean(streamsConfig);
}
----
====
IMPORTANT: Starting with version 2.2, the stream configuration is now provided as a `KafkaStreamsConfiguration` object, rather than a `StreamsConfig`.
IMPORTANT: Starting with version 2.2, the stream configuration is now provided as a `KafkaStreamsConfiguration` object rather than a `StreamsConfig`.
The `StreamsBuilderFactoryBean` also implements `SmartLifecycle` to manage lifecycle of an internal `KafkaStreams` instance.
Similar to the Kafka Streams API, the `KStream` instances must be defined before starting the `KafkaStreams`, and that also applies for the Spring API for Kafka Streams.
Therefore we have to declare `KStream` s on the `StreamsBuilder` before the application context is refreshed, when we use default `autoStartup = true` on the `StreamsBuilderFactoryBean`.
For example, `KStream` can be just as a regular bean definition, meanwhile the Kafka Streams API is used without any impacts:
The `StreamsBuilderFactoryBean` also implements `SmartLifecycle` to manage the lifecycle of an internal `KafkaStreams` instance.
Similar to the Kafka Streams API, you must define the `KStream` instances before you start the `KafkaStreams`.
That also applies for the Spring API for Kafka Streams.
Therefore, when you use default `autoStartup = true` on the `StreamsBuilderFactoryBean`, you must declare `KStream` instances on the `StreamsBuilder` before the application context is refreshed.
For example, `KStream` can be a regular bean definition, while the Kafka Streams API is used without any impacts.
The following example shows how to do so:
====
[source, java]
----
@Bean
@@ -67,17 +78,20 @@ public KStream<?, ?> kStream(StreamsBuilder kStreamBuilder) {
return stream;
}
----
====
If you would like to control lifecycle manually (e.g. stop and start by some condition), you can reference the `StreamsBuilderFactoryBean` bean directly using factory bean (`&`) http://docs.spring.io/spring/docs/current/spring-framework-reference/html/beans.html#beans-factory-extension-factorybean[prefix].
Since `StreamsBuilderFactoryBean` utilize its internal `KafkaStreams` instance, it is safe to stop and restart it again - a new `KafkaStreams` is created on each `start()`.
Also consider using different `StreamsBuilderFactoryBean` s, if you would like to control lifecycles for `KStream` instances separately.
If you would like to control the lifecycle manually (for example, stopping and starting by some condition), you can reference the `StreamsBuilderFactoryBean` bean directly by using the factory bean (`&`) http://docs.spring.io/spring/docs/current/spring-framework-reference/html/beans.html#beans-factory-extension-factorybean[prefix].
Since `StreamsBuilderFactoryBean` use its internal `KafkaStreams` instance, it is safe to stop and restart it again.
A new `KafkaStreams` is created on each `start()`.
You might also consider using different `StreamsBuilderFactoryBean` instances, if you would like to control the lifecycles for `KStream` instances separately.
You also can specify `KafkaStreams.StateListener`, `Thread.UncaughtExceptionHandler` and `StateRestoreListener` options on the `StreamsBuilderFactoryBean` which are delegated to the internal `KafkaStreams` instance.
Also apart from setting those options indirectly on `StreamsBuilderFactoryBean`, starting with _version 2.1.5_, a `KafkaStreamsCustomizer` callback interface can be used to configure inner `KafkaStreams` instance.
Note that `KafkaStreamsCustomizer` will override the options which are given via `StreamsBuilderFactoryBean`.
That internal `KafkaStreams` instance can be accessed via `StreamsBuilderFactoryBean.getKafkaStreams()` if you need to perform some `KafkaStreams` operations directly.
You can autowire `StreamsBuilderFactoryBean` bean by type, but you should be sure that you use full type in the bean definition, for example:
You also can specify `KafkaStreams.StateListener`, `Thread.UncaughtExceptionHandler`, and `StateRestoreListener` options on the `StreamsBuilderFactoryBean`, which are delegated to the internal `KafkaStreams` instance.
Also, apart from setting those options indirectly on `StreamsBuilderFactoryBean`, starting with _version 2.1.5_, you can use a `KafkaStreamsCustomizer` callback interface to configure an inner `KafkaStreams` instance.
Note that `KafkaStreamsCustomizer` overrides the options provided by `StreamsBuilderFactoryBean`.
If you need to perform some `KafkaStreams` operations directly, you can access that internal `KafkaStreams` instance by using `StreamsBuilderFactoryBean.getKafkaStreams()`.
You can autowire `StreamsBuilderFactoryBean` bean by type, but you should be sure to use the full type in the bean definition, as the following example shows:
====
[source,java]
----
@Bean
@@ -88,8 +102,12 @@ public StreamsBuilderFactoryBean myKStreamBuilder(KafkaStreamsConfiguration stre
@Autowired
private StreamsBuilderFactoryBean myKStreamBuilderFactoryBean;
----
====
Or add `@Qualifier` for injection by name if you use interface bean definition:
Alternatively, you can add `@Qualifier` for injection by name if you use interface bean definition.
The following example shows how to do so:
====
[source,java]
----
@Bean
@@ -101,29 +119,33 @@ public FactoryBean<StreamsBuilder> myKStreamBuilder(KafkaStreamsConfiguration st
@Qualifier("&myKStreamBuilder")
private StreamsBuilderFactoryBean myKStreamBuilderFactoryBean;
----
====
==== JSON Serdes
==== JSON Serialization and Deserialization
For serializing and deserializing data when reading or writing to topics or state stores in JSON format, Spring Kafka provides a `JsonSerde` implementation using JSON, delegating to the `JsonSerializer` and `JsonDeserializer` described in <<serdes, the serialization/deserialization section>>.
The `JsonSerde` provides the same configuration options via its constructor (target type and/or `ObjectMapper`).
In the following example we use the `JsonSerde` to serialize and deserialize the `Foo` payload of a Kafka stream - the `JsonSerde` can be used in a similar fashion wherever an instance is required.
For serializing and deserializing data when reading or writing to topics or state stores in JSON format, Spring Kafka provides a `JsonSerde` implementation that uses JSON, delegating to the `JsonSerializer` and `JsonDeserializer` described in <<serdes>>.
The `JsonSerde` implementation provides the same configuration options through its constructor (target type or `ObjectMapper`).
In the following example, we use the `JsonSerde` to serialize and deserialize the `Cat` payload of a Kafka stream (the `JsonSerde` can be used in a similar fashion wherever an instance is required):
====
[source,java]
----
stream.through(Serdes.Integer(), new JsonSerde<>(Foo.class), "foos");
stream.through(Serdes.Integer(), new JsonSerde<>(Cat.class), "cats");
----
====
IMPORTANT: Since Kafka Streams do not support headers, the `addTypeInfo` property on the `JsonSerializer` is ignored.
==== KafkaStreamsBrancher
==== Using `KafkaStreamsBrancher`
`KafkaStreamBrancher` class introduces more convenient way to build conditional branches on top of `KStream`.
The `KafkaStreamBrancher` class introduces a more convenient way to build conditional branches on top of `KStream`.
Instead of
Consider the following example that does not use `KafkaStreamBrancher`:
====
[source,java]
----
KStream<String, String>[] branches = builder.stream("source").branch(
KStream<String, String>[] branches = builder.stream("source").branch(
(key, value) -> value.contains("A"),
(key, value) -> value.contains("B"),
(key, value) -> true
@@ -132,9 +154,11 @@ branches[0].to("A");
branches[1].to("B");
branches[2].to("C");
----
====
we can use
The following example uses `KafkaStreamBrancher`:
====
[source,java]
----
new KafkaStreamsBrancher<String, String>()
@@ -145,27 +169,29 @@ new KafkaStreamsBrancher<String, String>()
.onTopOf(builder.stream("source"));
//onTopOf method returns the provided stream so we can continue with method chaining
----
====
[[streams-config]]
==== Configuration
To configure the Kafka Streams environment, the `StreamsBuilderFactoryBean` requires a `KafkaStreamsConfiguration` instance.
See Apache Kafka https://kafka.apache.org/0102/documentation/#streamsconfigs[documentation] for all possible options.
See the Apache Kafka https://kafka.apache.org/0102/documentation/#streamsconfigs[documentation] for all possible options.
IMPORTANT: Starting with version 2.2, the stream configuration is now provided as a `KafkaStreamsConfiguration` object, rather than a `StreamsConfig`.
IMPORTANT: Starting with version 2.2, the stream configuration is now provided as a `KafkaStreamsConfiguration` object, rather than as a `StreamsConfig`.
To avoid boilerplate code for most cases, especially when you develop micro services, Spring for Apache Kafka provides the `@EnableKafkaStreams` annotation, which should be placed on a `@Configuration` class.
All you need is to declare a `KafkaStreamsConfiguration` bean with the name `defaultKafkaStreamsConfig`.
A `StreamsBuilder` bean, with the name `defaultKafkaStreamsBuilder`, will be declared in the application context automatically.
Any additional `StreamsBuilderFactoryBean` beans can be declared and used as well.
To avoid boilerplate code for most cases, especially when you develop microservices, Spring for Apache Kafka provides the `@EnableKafkaStreams` annotation, which you should placed on a `@Configuration` class.
All you need is to declare a `KafkaStreamsConfiguration` bean named `defaultKafkaStreamsConfig`.
A `StreamsBuilder` bean, named `defaultKafkaStreamsBuilder`, is automatically declared in the application context.
You can declare and use any additional `StreamsBuilderFactoryBean` beans as well.
By default, when the factory bean is stopped, the `KafkaStreams.cleanUp()` method is called.
Starting with _version 2.1.2_, the factory bean has additional constructors, taking a `CleanupConfig` object that has properties to allow you to control whether the `cleanUp()` method is called during `start()`, `stop()`, or neither.
Starting with version 2.1.2, the factory bean has additional constructors, taking a `CleanupConfig` object that has properties to let you control whether the `cleanUp()` method is called during `start()` or `stop()` or neither.
==== Kafka Streams Example
Putting it all together:
The following example combines all the topics we have covered in this chapter:
====
[source, java]
----
@Configuration
@@ -204,3 +230,4 @@ public static class KafkaStreamsConfig {
}
----
====

114
src/reference/asciidoc/testing.adoc
View File

@@ -1,14 +1,14 @@
[[testing]]
=== Testing Applications
==== Introduction
The `spring-kafka-test` jar contains some useful utilities to assist with testing your applications.
==== JUnit
`o.s.kafka.test.utils.KafkaTestUtils` provides some static methods to set up producer and consumer properties:
`o.s.kafka.test.utils.KafkaTestUtils` provides some static methods to set up producer and consumer properties.
The following listing shows those method signatures:
====
[source, java]
----
/**
@@ -28,10 +28,13 @@ public static Map<String, Object> consumerProps(String group, String autoCommit,
*/
public static Map<String, Object> senderProps(EmbeddedKafkaBroker embeddedKafka) { ... }
----
====
A JUnit 4 `@Rule` wrapper for the `EmbeddedKafkaBroker` is provided that creates an embedded Kafka and an embedded Zookeeper server.
(See <<embedded-kafka-annotation>> about using `@EmbeddedKafka` with JUnit 5).
A JUnit 4 `@Rule` wrapper for the `EmbeddedKafkaBroker` is provided to create an embedded Kafka and an embedded Zookeeper server.
(See <<embedded-kafka-annotation>> for information about using `@EmbeddedKafka` with JUnit 5).
The following listing shows the signatures of those methods:
====
[source, java]
----
/**
@@ -52,9 +55,12 @@ public EmbeddedKafkaRule(int count, boolean controlledShutdown, String... topics
*/
public EmbeddedKafkaRule(int count, boolean controlledShutdown, int partitions, String... topics) { ... }
----
====
The `EmbeddedKafkaBroker` class has a utility method allowing you to consume for all the topics it created:
The `EmbeddedKafkaBroker` class has a utility method that lets you consume for all the topics it created.
The following example shows how to use it:
====
[source, java]
----
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testT", "false", embeddedKafka);
@@ -63,9 +69,12 @@ DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactor
Consumer<Integer, String> consumer = cf.createConsumer();
embeddedKafka.consumeFromAllEmbeddedTopics(consumer);
----
====
The `KafkaTestUtils` has some utility methods to fetch results from the consumer:
The `KafkaTestUtils` has some utility methods to fetch results from the consumer.
The following listing shows those method signatures:
====
[source, java]
----
/**
@@ -84,9 +93,11 @@ public static <K, V> ConsumerRecord<K, V> getSingleRecord(Consumer<K, V> consume
*/
public static <K, V> ConsumerRecords<K, V> getRecords(Consumer<K, V> consumer) { ... }
----
====
Usage:
The following example shows how to use `KafkaTestUtils`:
====
[source, java]
----
...
@@ -94,27 +105,30 @@ template.sendDefault(0, 2, "bar");
ConsumerRecord<Integer, String> received = KafkaTestUtils.getSingleRecord(consumer, "topic");
...
----
====
When the embedded Kafka and embedded Zookeeper server are started by by the `EmbeddedKafkaBroker`, a system property `spring.embedded.kafka.brokers` is set to the address of the Kafka broker(s) and a system property `spring.embedded.zookeeper.connect` is set to the address of Zookeeper.
Convenient constants `EmbeddedKafkaBroker.SPRING_EMBEDDED_KAFKA_BROKERS` and `EmbeddedKafkaBroker.SPRING_EMBEDDED_ZOOKEEPER_CONNECT` are provided for this property.
When the embedded Kafka and embedded Zookeeper server are started by the `EmbeddedKafkaBroker`, a system property named `spring.embedded.kafka.brokers` is set to the address of the Kafka brokers and a system property named `spring.embedded.zookeeper.connect` is set to the address of Zookeeper.
Convenient constants (`EmbeddedKafkaBroker.SPRING_EMBEDDED_KAFKA_BROKERS` and `EmbeddedKafkaBroker.SPRING_EMBEDDED_ZOOKEEPER_CONNECT`) are provided for this property.
With the `EmbeddedKafkaBroker.brokerProperties(Map<String, String>)` you can provide additional properties for the Kafka server(s).
With the `EmbeddedKafkaBroker.brokerProperties(Map<String, String>)`, you can provide additional properties for the Kafka servers.
See https://kafka.apache.org/documentation/#brokerconfigs[Kafka Config] for more information about possible broker properties.
==== Configuring Topics
The following example configuration creates topics called `cat` and `hat` with five partitions, a topic called `thing1` with 10 partitions, and a topic called `thing2` with 15 partitions:
====
[source, java]
----
public class MyTests {
@ClassRule
private static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, false, 5, "foo", "bar");
private static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, false, 5, "cat", "hat");
@Test
public void test() {
embeddedKafkaRule.getEmbeddedKafka()
.addTopics(new NewTopic("baz", 10, (short) 1), new NewTopic("qux", 15, (short) 1));
.addTopics(new NewTopic("thing1", 10, (short) 1), new NewTopic("thing2", 15, (short) 1));
...
}
@@ -122,11 +136,9 @@ public class MyTests {
----
====
The above configuration will create topics `foo` and `bar` with 5 partitions, `baz` with 10 and `qux` with 15.
==== Using the Same Brokers for Multiple Test Classes
==== Using the Same Broker(s) for Multiple Test Classes
There is no built-in support for this, but it can be achieved with something similar to the following:
There is no built-in support for doing so, but you can use the same broker for multiple test classes with something similar to the following:
====
[source, java]
@@ -158,7 +170,7 @@ public final class EmbeddedKafkaHolder {
----
====
And then, in each test class:
Then, in each test class, you can use something similar to the following:
====
[source, java]
@@ -171,16 +183,18 @@ private static EmbeddedKafkaRule embeddedKafka = EmbeddedKafkaHolder.getEmbedded
----
====
IMPORTANT: This example provides no mechanism for shutting down the broker(s) when all tests are complete.
IMPORTANT: The preceding example provides no mechanism for shutting down the brokers when all tests are complete.
This could be a problem if, say, you run your tests in a Gradle daemon.
You should not use this technique in such a situation, or use something to call `destroy()` on the `EmbeddedKafkaBroker` when your tests are complete.
You should not use this technique in such a situation, or you should use something to call `destroy()` on the `EmbeddedKafkaBroker` when your tests are complete.
[[embedded-kafka-annotation]]
==== @EmbeddedKafka Annotation
It is generally recommended to use the rule as a `@ClassRule` to avoid starting/stopping the broker between tests (and use a different topic for each test).
Starting with _version 2.0_, if you are using Spring's test application context caching, you can also declare a `EmbeddedKafkaBroker` bean, so a single broker can be used across multiple test classes.
For convenience a test class level `@EmbeddedKafka` annotation is provided with the purpose to register `EmbeddedKafkaBroker` bean:
We generally recommend that you use the rule as a `@ClassRule` to avoid starting and stopping the broker between tests (and use a different topic for each test).
Starting with version 2.0, if you use Spring's test application context caching, you can also declare a `EmbeddedKafkaBroker` bean, so a single broker can be used across multiple test classes.
For convenience, we provide a test class-level annotation called `@EmbeddedKafka` to register the `EmbeddedKafkaBroker` bean.
The following example shows how to use it:
====
[source, java]
----
@RunWith(SpringRunner.class)
@@ -224,9 +238,13 @@ public class KafkaStreamsTests {
}
----
Starting with version 2.2.4, the `@EmbeddedKafka` annotation can also be used to specify the kafka ports property.
====
The `topics`, `brokerProperties` and `brokerPropertiesLocation` attributes of `@EmbeddedKafka` support property placeholder resolutions:
Starting with version 2.2.4, you can also use the `@EmbeddedKafka` annotation to specify the Kafka ports property.
The following example sets the `topics`, `brokerProperties`, and `brokerPropertiesLocation` attributes of `@EmbeddedKafka` support property placeholder resolutions:
====
[source, java]
----
@TestPropertySource(locations = "classpath:/test.properties")
@@ -236,21 +254,26 @@ The `topics`, `brokerProperties` and `brokerPropertiesLocation` attributes of `@
"auto.create.topics.enable=${kafka.broker.topics-enable:true}" }
brokerPropertiesLocation = "classpath:/broker.properties")
----
In the example above, the property placeholders `${kafka.topics.another-topic}`, `${kafka.broker.logs-dir}`, and `${kafka.broker.port}` are resolved from the Spring `Environment`.
In addition the broker properties are loaded from the `broker.properties` classpath resource specified by the `brokerPropertiesLocation`.
====
In the preceding example, the property placeholders `${kafka.topics.another-topic}`, `${kafka.broker.logs-dir}`, and `${kafka.broker.port}` are resolved from the Spring `Environment`.
In addition, the broker properties are loaded from the `broker.properties` classpath resource specified by the `brokerPropertiesLocation`.
Property placeholders are resolved for the `brokerPropertiesLocation` URL and for any property placeholders found in the resource.
Properties defined by `brokerProperties` override properties found in `brokerPropertiesLocation`.
The `@EmbeddedKafka` annotation can be used with JUnit 4 or JUnit 5.
You can use the `@EmbeddedKafka` annotation with JUnit 4 or JUnit 5.
==== Embedded Broker in `@SpringBootTest` s
==== Embedded Broker in `@SpringBootTest` Annotations
https://start.spring.io/[Spring Initializr] now automatically adds the `spring-kafka-test` dependency in test scope to the project configuration.
IMPORTANT: If your application is using the Kafka binder in `spring-cloud-stream`, if you want to use an embedded broker for tests, you must remove the `spring-cloud-stream-test-support` dependency because it replaces the real binder with a test binder for test casess.
If you wish some tests to use the test binder and some to use the embedded broker, tests using the real binder need to disable the test binder by excluding the binder auto configuration in the test class.
[IMPORTANT]
====
If your application uses the Kafka binder in `spring-cloud-stream` and if you want to use an embedded broker for tests, you must remove the `spring-cloud-stream-test-support` dependency, because it replaces the real binder with a test binder for test cases.
If you wish some tests to use the test binder and some to use the embedded broker, tests that use the real binder need to disable the test binder by excluding the binder auto configuration in the test class.
The following example shows how to do so:
=====
[source, java]
----
@RunWith(SpringRunner.class)
@@ -260,12 +283,21 @@ public class MyApplicationTests {
...
}
----
=====
====
There are several ways to use an embedded broker in a Spring Boot application test.
They include:
* <<kafka-testing-junit4-class-rule>>
* <<kafka-testing-embeddedkafka-annotation>>
[[kafka-testing-junit4-class-rule]]
===== JUnit4 Class Rule
The following example shows how to use a JUnit4 class rule to create an embedded broker:
====
[source, java]
----
@@ -295,8 +327,11 @@ public class MyApplicationTests {
----
====
[[kafka-testing-embeddedkafka-annotation]]
===== `@EmbeddedKafka` Annotation or `EmbeddedKafkaBroker` Bean
The following example shows how to use an `@EmbeddedKafka` Annotation to create an embedded broker:
====
[source, java]
----
@@ -326,6 +361,7 @@ public class MyApplicationTests {
The `o.s.kafka.test.hamcrest.KafkaMatchers` provides the following matchers:
====
[source, java]
----
/**
@@ -369,9 +405,13 @@ public static Matcher<ConsumerRecord<?, ?>> hasTimestamp(TimestampType type, lon
return new ConsumerRecordTimestampMatcher(type, ts);
}
----
====
==== AssertJ Conditions
You can use the following AssertJ conditions:
====
[source, java]
----
/**
@@ -411,11 +451,13 @@ public static Condition<ConsumerRecord<?, ?>> timestamp(TimestampType type, long
return new ConsumerRecordTimestampCondition(type, value);
}
----
====
==== Example
Putting it all together:
The following example brings together most of the topics covered in this chapter:
====
[source, java]
----
public class KafkaTemplateTests {
@@ -469,9 +511,12 @@ public class KafkaTemplateTests {
}
----
====
The above uses the hamcrest matchers; with `AssertJ`, the final part looks like this...
The preceding example uses the Hamcrest matchers.
With `AssertJ`, the final part looks like the following code:
====
[source, java]
----
assertThat(records.poll(10, TimeUnit.SECONDS)).has(value("foo"));
@@ -486,3 +531,4 @@ assertThat(received).has(key(2));
assertThat(received).has(partition(0));
assertThat(received).has(value("baz"));
----
====

59
src/reference/asciidoc/whats-new.adoc
View File

@@ -1,46 +1,46 @@
=== What's new in 2.2 Since 2.1
=== What's New in 2.2 Since 2.1
This section covers the changes made from version 2.1 to version 2.2.
==== Kafka Client Version
This version requires the 2.0.0 `kafka-clients` or higher.
==== Class/Package Changes
==== Class and Package Changes
The class `ContainerProperties` has been moved from `org.springframework.kafka.listener.config` to `org.springframework.kafka.listener`.
The `ContainerProperties` class has been moved from `org.springframework.kafka.listener.config` to `org.springframework.kafka.listener`.
The enum `AckMode` has been moved from `AbstractMessageListenerContainer` to `ContainerProperties`.
The `AckMode` enum has been moved from `AbstractMessageListenerContainer` to `ContainerProperties`.
`setBatchErrorHandler()` and `setErrorHandler()` methods have been moved from `ContainerProperties` to `AbstractMessageListenerContainer` (and `AbstractKafkaListenerContainerFactory`).
The `setBatchErrorHandler()` and `setErrorHandler()` methods have been moved from `ContainerProperties` to both `AbstractMessageListenerContainer` and `AbstractKafkaListenerContainerFactory`.
==== After rollback processing
==== After Rollback Processing
A new `AfterRollbackProcessor` strategy is provided - see <<after-rollback>> for more information.
A new `AfterRollbackProcessor` strategy is provided.
See <<after-rollback>> for more information.
==== ConcurrentKafkaListenerContainerFactory changes
==== `ConcurrentKafkaListenerContainerFactory` Changes
The `ConcurrentKafkaListenerContainerFactory` can now be used to create/configure any `ConcurrentMessageListenerContainer`, not just those for `@KafkaListener` annotations.
You can now use the `ConcurrentKafkaListenerContainerFactory` to create and configure any `ConcurrentMessageListenerContainer`, not only those for `@KafkaListener` annotations.
See <<container-factory>> for more information.
==== Listener Container Changes
A new container property `missingTopicsFatal` has been added.
A new container property (`missingTopicsFatal`) has been added.
See <<kafka-container>> for more information.
A `ConsumerStoppedEvent` is now emitted when a consumer terminates.
See <<thread-safety>> for more information.
Batch listeners can optionally receive the complete `ConsumerRecords<?, ?>` object instead of a `List<ConsumerRecord<?, ?>`.
See <<batch-listeners>> for more information.
The `DefaultAfterRollbackProcessor` and `SeekToCurrentErrorHandler` can now recover (skip) records that keep failing, and will do so after 10 failures, by default.
The `DefaultAfterRollbackProcessor` and `SeekToCurrentErrorHandler` can now recover (skip) records that keep failing, and, by default, does so after 10 failures.
They can be configured to publish failed records to a dead-letter topic.
Starting with version 2.2.4, the consumer's group id can be used while selecting the dead letter topic name.
Starting with version 2.2.4, the consumer's group ID can be used while selecting the dead letter topic name.
See <<after-rollback>>, <<seek-to-current>> and <<dead-letters>> for more information.
See <<after-rollback>>, <<seek-to-current>>, and <<dead-letters>> for more information.
The `ConsumerStoppingEvent` has been added.
See <<events>> for more information.
@@ -49,11 +49,9 @@ See <<events>> for more information.
You can now override the `concurrency` and `autoStartup` properties of the listener container factory by setting properties on the annotation.
You can now add configuration to determine which headers (if any) are copied to a reply message.
See <<kafka-listener-annotation>> for more information.
You can now use `@KafkaListener` as a meta-annotation on your own annotations.
See <<kafka-listener-meta>> for more information.
It is now easier to configure a `Validator` for `@Payload` validation.
@@ -62,37 +60,36 @@ See <<kafka-validation>> for more information.
==== Header Mapping Changes
Headers of type `MimeType` and `MediaType` are now mapped as simple strings in the `RecordHeader` value.
Previously, they were mapped as JSON and only `MimeType` was decoded, `MediaType` could not be decoded.
Previously, they were mapped as JSON and only `MimeType` was decoded.
`MediaType` could not be decoded.
They are now simple strings for interoperability.
Also, the `DefaultKafkaHeaderMapper` has a new method `addToStringClasses` allowing the specification of types that should be mapped using `toString()` instead of JSON.
Also, the `DefaultKafkaHeaderMapper` has a new `addToStringClasses` method, allowing the specification of types that should be mapped by using `toString()` instead of JSON.
See <<headers>> for more information.
==== Embedded Kafka Changes
The `KafkaEmbedded` class and its `KafkaRule` interface have need deprecated in favor of the `EmbeddedKafkaBroker` and its JUnit 4 `EmbeddedKafkaRule` wrapper.
The `KafkaEmbedded` class and its `KafkaRule` interface have been deprecated in favor of the `EmbeddedKafkaBroker` and its JUnit 4 `EmbeddedKafkaRule` wrapper.
The `@EmbeddedKafka` annotation now populates an `EmbeddedKafkaBroker` bean instead of the deprecated `KafkaEmbedded`.
This allows the use of `@EmbeddedKafka` in JUnit 5 tests.
The `@EmbeddedKafka` annotation now has the attribute `ports` to specify the port which will populate the `EmbeddedKafkaBroker`.
This change allows the use of `@EmbeddedKafka` in JUnit 5 tests.
The `@EmbeddedKafka` annotation now has the attribute `ports` to specify the port that populates the `EmbeddedKafkaBroker`.
See <<testing>> for more information.
==== JsonSerializer/Deserializer Enhancements
You can now provide type mapping information using producer/consumer properties.
You can now provide type mapping information by using producer and consumer properties.
New constructors are available on the deserializer to allow overriding the type header information with the supplied target type.
The `JsonDeserializer` will now remove any type information headers by default.
The `JsonDeserializer` now removes any type information headers by default.
The `JsonDeserializer` can now be configured to ignore type information headers using a kafka property (since 2.2.3).
You can now configure the `JsonDeserializer` to ignore type information headers by using a Kafka property (since 2.2.3).
See <<serdes>> for more information.
==== Kafka Streams Changes
The streams configuration bean must now be a `KafkaStreamsConfiguration` object instead of a `StreamsConfig`.
The streams configuration bean must now be a `KafkaStreamsConfiguration` object instead of a `StreamsConfig` object.
The `StreamsBuilderFactoryBean` has been moved from package `...core` to `...config`.
@@ -101,7 +98,7 @@ The `KafkaStreamBrancher` has been introduced for better end-user experience whe
See <<kafka-streams>> and <<streams-config>> for more information.
==== Transactional Id
==== Transactional ID
When a transaction is started by the listener container, the `transactional.id` is now the `transactionIdPrefix` appended with `<group.id>.<topic>.<partition>`.
This is to allow proper fencing of zombies https://www.confluent.io/blog/transactions-apache-kafka/[as described here].
This change allows proper fencing of zombies, https://www.confluent.io/blog/transactions-apache-kafka/[as described here].