GH-1216: Move Transactions to the top level

Fixes #1216

It is hard to determine transactions support in the reference manual
because it is hidden under `Sending Messages`

* Move `Transactions` to the top level and after `Receiving Messages`
* Upgrade to Spring Data Moore RC3; Gradle 5.6.2

Author: artembilan

build.gradle

@@ -54,7 +54,7 @@ ext {
 	scalaVersion = '2.12'
 	springRetryVersion = '1.2.4.RELEASE'
 	springVersion = '5.2.0.RC2'
-	springDataCommonsVersion = '2.2.0.BUILD-SNAPSHOT'
+	springDataCommonsVersion = '2.2.0.RC3'
 
 	idPrefix = 'kafka'
 }

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-5.6.1-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-5.6.2-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists

src/reference/asciidoc/kafka.adoc

@@ -321,155 +321,9 @@ public KafkaTemplate<Integer, CustomValue> kafkaTemplate() {
     return new KafkaTemplate<Integer, CustomValue>(producerFactory());
 }
 
-----
-====
-[[transactions]]
-===== Transactions
-
-This section describes how Spring for Apache Kafka supports transactions.
-
-====== Overview
-
-The 0.11.0.0 client library added support for transactions.
-Spring for Apache Kafka adds support in the following ways:
-
-* `KafkaTransactionManager`: Used with normal Spring transaction support (`@Transactional`, `TransactionTemplate` etc).
-* Transactional `KafkaMessageListenerContainer`
-* Local transactions with `KafkaTemplate`
-
-Transactions are enabled by providing the `DefaultKafkaProducerFactory` with a `transactionIdPrefix`.
-In that case, instead of managing a single shared `Producer`, the factory maintains a cache of transactional producers.
-When the user calls `close()` on a producer, it is returned to the cache for reuse instead of actually being closed.
-The `transactional.id` property of each producer is `transactionIdPrefix` + `n`, where `n` starts with `0` and is incremented for each new producer, unless the transaction is started by a listener container with a record-based listener.
-In that case, the `transactional.id` is `<transactionIdPrefix>.<group.id>.<topic>.<partition>`.
-This is to properly support fencing zombies, https://www.confluent.io/blog/transactions-apache-kafka/[as described here].
-This new behavior was added in versions 1.3.7, 2.0.6, 2.1.10, and 2.2.0.
-If you wish to revert to the previous behavior, you can set the `producerPerConsumerPartition` property on the `DefaultKafkaProducerFactory` to `false`.
-
-NOTE: While transactions are supported with batch listeners, zombie fencing cannot be supported because a batch may contain records from multiple topics or partitions.
-
-Also see <<transaction-id-prefix>>.
-
-====== Using `KafkaTransactionManager`
-
-The `KafkaTransactionManager` is an implementation of Spring Framework's `PlatformTransactionManager`.
-It is provided with a reference to the producer factory in its constructor.
-If you provide a custom producer factory, it must support transactions.
-See `ProducerFactory.transactionCapable()`.
-
-You can use the `KafkaTransactionManager` with normal Spring transaction support (`@Transactional`, `TransactionTemplate`, and others).
-If a transaction is active, any `KafkaTemplate` operations performed within the scope of the transaction use the transaction's `Producer`.
-The manager commits or rolls back the transaction, depending on success or failure.
-You must configure the `KafkaTemplate` to use the same `ProducerFactory` as the transaction manager.
-
-====== Transactional Listener Container and Exactly Once Processing
-
-You can provide a listener container with a `KafkaAwareTransactionManager` instance.
-When so configured, the container starts a transaction before invoking the listener.
-Any `KafkaTemplate` operations performed by the listener participate in the transaction.
-If the listener successfully processes the record (or multiple records, when using a `BatchMessageListener`), the container sends the offsets to the transaction by using `producer.sendOffsetsToTransaction()`), before the transaction manager commits the transaction.
-If the listener throws an exception, the transaction is rolled back and the consumer is repositioned so that the rolled-back record(s) can be retrieved on the next poll.
-See <<after-rollback>> for more information and for handling records that repeatedly fail.
-
-====== Transaction Synchronization
-
-If you need to synchronize a Kafka transaction with some other transaction, configure the listener container with the appropriate transaction manager (one that supports synchronization, such as the `DataSourceTransactionManager`).
-Any operations performed on a transactional `KafkaTemplate` from the listener participate in a single transaction.
-The Kafka transaction is committed (or rolled back) immediately after the controlling transaction.
-Before exiting the listener, you should invoke one of the template's `sendOffsetsToTransaction` methods (unless you use a <<chained-transaction-manager,`ChainedKafkaTransactionManager`>>).
-For convenience, the listener container binds its consumer group ID to the thread, so, generally, you can use the first method.
-The following listing shows the two method signatures:
-
-====
-[source, java]
-----
-void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets);
-
-void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId);
-----
-====
-
-The following example shows how to use the first signature of the `sendOffsetsToTransaction` method:
-
-====
-[source, java]
-----
-@Bean
-KafkaMessageListenerContainer container(ConsumerFactory<String, String> cf,
-            final KafkaTemplate template) {
-    ContainerProperties props = new ContainerProperties("foo");
-    props.setGroupId("group");
-    props.setTransactionManager(new SomeOtherTransactionManager());
-    ...
-    props.setMessageListener((MessageListener<String, String>) m -> {
-        template.send("foo", "bar");
-        template.send("baz", "qux");
-        template.sendOffsetsToTransaction(
-            Collections.singletonMap(new TopicPartition(m.topic(), m.partition()),
-                new OffsetAndMetadata(m.offset() + 1)));
-    });
-    return new KafkaMessageListenerContainer<>(cf, props);
-}
-----
-====
-
-NOTE: The offset to be committed is one greater than the offset of the records processed by the listener.
-
-IMPORTANT: You should call this only when you use transaction synchronization.
-When a listener container is configured to use a `KafkaTransactionManager` or `ChainedKafkaTransactionManager`, it takes care of sending the offsets to the transaction.
-
-See <<ex-jdbc-sync>> for an example application that synchronizes JDBC and Kafka transactions.
-
-[[chained-transaction-manager]]
-====== Using `ChainedKafkaTransactionManager`
-
-The `ChainedKafkaTransactionManager` was introduced in version 2.1.3.
-This is a subclass of `ChainedTransactionManager` that can have exactly one `KafkaTransactionManager`.
-Since it is a `KafkaAwareTransactionManager`, the container can send the offsets to the transaction in the same way as when the container is configured with a simple `KafkaTransactionManager`.
-This provides another mechanism for synchronizing transactions without having to send the offsets to the transaction in the listener code.
-You should chain your transaction managers in the desired order and provide the `ChainedTransactionManager` in the `ContainerProperties`.
-
-See <<ex-jdbc-sync>> for an example application that synchronizes JDBC and Kafka transactions.
-
-====== `KafkaTemplate` Local Transactions
-
-You can use the `KafkaTemplate` to execute a series of operations within a local transaction.
-The following example shows how to do so:
-
-====
-[source, java]
-----
-boolean result = template.executeInTransaction(t -> {
-    t.sendDefault("thing1", "thing2");
-    t.sendDefault("cat", "hat");
-    return true;
-});
 ----
 ====
 
-The argument in the callback is the template itself (`this`).
-If the callback exits normally, the transaction is committed.
-If an exception is thrown, the transaction is rolled back.
-
-NOTE: If there is a `KafkaTransactionManager` (or synchronized) transaction in process, it is not used.
-Instead, a new "nested" transaction is used.
-
-[[transaction-id-prefix]]
-====== `transactionIdPrefix`
-
-As mentioned in <<transactions, the overview>>, the producer factory is configured with this property to build the producer `transactional.id` property.
-There is rather a dichotomy when specifying this property in that, when running multiple instances of the application, it must be the same on all instances to satisfy fencing zombies (also mentioned in the overview) when producing records on a listener container thread.
-However, when producing records using transactions that are **not** started by a listener container, the prefix has to be different on each instance.
-Version 2.3, makes this simpler to configure, especially in a Spring Boot application.
-In previous versions, you had to create two producer factories and `KafkaTemplate` s - one for producing records on a listener container thread and one for stand-alone transactions started by `kafkaTemplate.executeInTransaction()` or by a transaction interceptor on a `@Transactional` method.
-
-Now, you can override the factory's `transactionalIdPrefix` on the `KafkaTemplate` and the `KafkaTransactionManager`.
-
-When using a transaction manager and template for a listener container, you would normally leave this to default to the producer factory's property.
-This value should be the same for all application instances.
-For transactions started by the template (or the transaction manager for `@Transaction`) you should set the property on the template and transaction manager respectively.
-This property must have a different value on each application instance.
-
 [[replying-template]]
 ===== Using `ReplyingKafkaTemplate`
 
@@ -2298,6 +2152,153 @@ Note that `SimpleThreadScope` does not destroy beans that have a destruction int
 IMPORTANT: By default, the application context's event multicaster invokes event listeners on the calling thread.
 If you change the multicaster to use an async executor, thread cleanup is not effective.
 
+[[transactions]]
+==== Transactions
+
+This section describes how Spring for Apache Kafka supports transactions.
+
+===== Overview
+
+The 0.11.0.0 client library added support for transactions.
+Spring for Apache Kafka adds support in the following ways:
+
+* `KafkaTransactionManager`: Used with normal Spring transaction support (`@Transactional`, `TransactionTemplate` etc).
+* Transactional `KafkaMessageListenerContainer`
+* Local transactions with `KafkaTemplate`
+
+Transactions are enabled by providing the `DefaultKafkaProducerFactory` with a `transactionIdPrefix`.
+In that case, instead of managing a single shared `Producer`, the factory maintains a cache of transactional producers.
+When the user calls `close()` on a producer, it is returned to the cache for reuse instead of actually being closed.
+The `transactional.id` property of each producer is `transactionIdPrefix` + `n`, where `n` starts with `0` and is incremented for each new producer, unless the transaction is started by a listener container with a record-based listener.
+In that case, the `transactional.id` is `<transactionIdPrefix>.<group.id>.<topic>.<partition>`.
+This is to properly support fencing zombies, https://www.confluent.io/blog/transactions-apache-kafka/[as described here].
+This new behavior was added in versions 1.3.7, 2.0.6, 2.1.10, and 2.2.0.
+If you wish to revert to the previous behavior, you can set the `producerPerConsumerPartition` property on the `DefaultKafkaProducerFactory` to `false`.
+
+NOTE: While transactions are supported with batch listeners, zombie fencing cannot be supported because a batch may contain records from multiple topics or partitions.
+
+Also see <<transaction-id-prefix>>.
+
+===== Using `KafkaTransactionManager`
+
+The `KafkaTransactionManager` is an implementation of Spring Framework's `PlatformTransactionManager`.
+It is provided with a reference to the producer factory in its constructor.
+If you provide a custom producer factory, it must support transactions.
+See `ProducerFactory.transactionCapable()`.
+
+You can use the `KafkaTransactionManager` with normal Spring transaction support (`@Transactional`, `TransactionTemplate`, and others).
+If a transaction is active, any `KafkaTemplate` operations performed within the scope of the transaction use the transaction's `Producer`.
+The manager commits or rolls back the transaction, depending on success or failure.
+You must configure the `KafkaTemplate` to use the same `ProducerFactory` as the transaction manager.
+
+===== Transactional Listener Container and Exactly Once Processing
+
+You can provide a listener container with a `KafkaAwareTransactionManager` instance.
+When so configured, the container starts a transaction before invoking the listener.
+Any `KafkaTemplate` operations performed by the listener participate in the transaction.
+If the listener successfully processes the record (or multiple records, when using a `BatchMessageListener`), the container sends the offsets to the transaction by using `producer.sendOffsetsToTransaction()`), before the transaction manager commits the transaction.
+If the listener throws an exception, the transaction is rolled back and the consumer is repositioned so that the rolled-back record(s) can be retrieved on the next poll.
+See <<after-rollback>> for more information and for handling records that repeatedly fail.
+
+===== Transaction Synchronization
+
+If you need to synchronize a Kafka transaction with some other transaction, configure the listener container with the appropriate transaction manager (one that supports synchronization, such as the `DataSourceTransactionManager`).
+Any operations performed on a transactional `KafkaTemplate` from the listener participate in a single transaction.
+The Kafka transaction is committed (or rolled back) immediately after the controlling transaction.
+Before exiting the listener, you should invoke one of the template's `sendOffsetsToTransaction` methods (unless you use a <<chained-transaction-manager,`ChainedKafkaTransactionManager`>>).
+For convenience, the listener container binds its consumer group ID to the thread, so, generally, you can use the first method.
+The following listing shows the two method signatures:
+
+====
+[source, java]
+----
+void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets);
+
+void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId);
+----
+====
+
+The following example shows how to use the first signature of the `sendOffsetsToTransaction` method:
+
+====
+[source, java]
+----
+@Bean
+KafkaMessageListenerContainer container(ConsumerFactory<String, String> cf,
+            final KafkaTemplate template) {
+    ContainerProperties props = new ContainerProperties("foo");
+    props.setGroupId("group");
+    props.setTransactionManager(new SomeOtherTransactionManager());
+    ...
+    props.setMessageListener((MessageListener<String, String>) m -> {
+        template.send("foo", "bar");
+        template.send("baz", "qux");
+        template.sendOffsetsToTransaction(
+            Collections.singletonMap(new TopicPartition(m.topic(), m.partition()),
+                new OffsetAndMetadata(m.offset() + 1)));
+    });
+    return new KafkaMessageListenerContainer<>(cf, props);
+}
+----
+====
+
+NOTE: The offset to be committed is one greater than the offset of the records processed by the listener.
+
+IMPORTANT: You should call this only when you use transaction synchronization.
+When a listener container is configured to use a `KafkaTransactionManager` or `ChainedKafkaTransactionManager`, it takes care of sending the offsets to the transaction.
+
+See <<ex-jdbc-sync>> for an example application that synchronizes JDBC and Kafka transactions.
+
+[[chained-transaction-manager]]
+===== Using `ChainedKafkaTransactionManager`
+
+The `ChainedKafkaTransactionManager` was introduced in version 2.1.3.
+This is a subclass of `ChainedTransactionManager` that can have exactly one `KafkaTransactionManager`.
+Since it is a `KafkaAwareTransactionManager`, the container can send the offsets to the transaction in the same way as when the container is configured with a simple `KafkaTransactionManager`.
+This provides another mechanism for synchronizing transactions without having to send the offsets to the transaction in the listener code.
+You should chain your transaction managers in the desired order and provide the `ChainedTransactionManager` in the `ContainerProperties`.
+
+See <<ex-jdbc-sync>> for an example application that synchronizes JDBC and Kafka transactions.
+
+===== `KafkaTemplate` Local Transactions
+
+You can use the `KafkaTemplate` to execute a series of operations within a local transaction.
+The following example shows how to do so:
+
+====
+[source, java]
+----
+boolean result = template.executeInTransaction(t -> {
+    t.sendDefault("thing1", "thing2");
+    t.sendDefault("cat", "hat");
+    return true;
+});
+----
+====
+
+The argument in the callback is the template itself (`this`).
+If the callback exits normally, the transaction is committed.
+If an exception is thrown, the transaction is rolled back.
+
+NOTE: If there is a `KafkaTransactionManager` (or synchronized) transaction in process, it is not used.
+Instead, a new "nested" transaction is used.
+
+[[transaction-id-prefix]]
+===== `transactionIdPrefix`
+
+As mentioned in <<transactions, the overview>>, the producer factory is configured with this property to build the producer `transactional.id` property.
+There is rather a dichotomy when specifying this property in that, when running multiple instances of the application, it must be the same on all instances to satisfy fencing zombies (also mentioned in the overview) when producing records on a listener container thread.
+However, when producing records using transactions that are **not** started by a listener container, the prefix has to be different on each instance.
+Version 2.3, makes this simpler to configure, especially in a Spring Boot application.
+In previous versions, you had to create two producer factories and `KafkaTemplate` s - one for producing records on a listener container thread and one for stand-alone transactions started by `kafkaTemplate.executeInTransaction()` or by a transaction interceptor on a `@Transactional` method.
+
+Now, you can override the factory's `transactionalIdPrefix` on the `KafkaTemplate` and the `KafkaTransactionManager`.
+
+When using a transaction manager and template for a listener container, you would normally leave this to default to the producer factory's property.
+This value should be the same for all application instances.
+For transactions started by the template (or the transaction manager for `@Transaction`) you should set the property on the template and transaction manager respectively.
+This property must have a different value on each application instance.
+
 [[interceptors]]
 ==== Wiring Spring Beans into Producer/Consumer Interceptors