From 60f92c87fc00c695daececbaf53fb5be860a67a4 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 16 Feb 2018 12:57:05 -0500 Subject: [PATCH] Add ChainedKafkaTransactionManager Enable transaction synchronization without the need for user code to send the offset(s) to the transaction. --- .../KafkaMessageListenerContainer.java | 8 +-- .../ChainedKafkaTransactionManager.java | 58 +++++++++++++++++++ .../KafkaAwareTransactionManager.java | 39 +++++++++++++ .../transaction/KafkaTransactionManager.java | 5 +- .../listener/TransactionalContainerTests.java | 23 ++++++-- src/reference/asciidoc/kafka.adoc | 11 +++- src/reference/asciidoc/whats-new.adoc | 4 ++ 7 files changed, 137 insertions(+), 11 deletions(-) create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/transaction/ChainedKafkaTransactionManager.java create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/transaction/KafkaAwareTransactionManager.java diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java index 5c454dcc..69230b5c 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java @@ -62,7 +62,7 @@ import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.LogIfLevelEnabled; import org.springframework.kafka.support.TopicPartitionInitialOffset; import org.springframework.kafka.support.TopicPartitionInitialOffset.SeekPosition; -import org.springframework.kafka.transaction.KafkaTransactionManager; +import org.springframework.kafka.transaction.KafkaAwareTransactionManager; import org.springframework.scheduling.SchedulingAwareRunnable; import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; @@ -361,9 +361,9 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener private final PlatformTransactionManager transactionManager = this.containerProperties.getTransactionManager(); @SuppressWarnings("rawtypes") - private final KafkaTransactionManager kafkaTxManager = - this.transactionManager instanceof KafkaTransactionManager - ? ((KafkaTransactionManager) this.transactionManager) : null; + private final KafkaAwareTransactionManager kafkaTxManager = + this.transactionManager instanceof KafkaAwareTransactionManager + ? ((KafkaAwareTransactionManager) this.transactionManager) : null; private final TransactionTemplate transactionTemplate; diff --git a/spring-kafka/src/main/java/org/springframework/kafka/transaction/ChainedKafkaTransactionManager.java b/spring-kafka/src/main/java/org/springframework/kafka/transaction/ChainedKafkaTransactionManager.java new file mode 100644 index 00000000..0da7c82f --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/transaction/ChainedKafkaTransactionManager.java @@ -0,0 +1,58 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.transaction; + +import org.springframework.data.transaction.ChainedTransactionManager; +import org.springframework.kafka.core.ProducerFactory; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.util.Assert; + +/** + * A {@link ChainedTransactionManager} that has exactly one + * {@link KafkaAwareTransactionManager} in the chain. + * + * @param the key type. + * @param the value type. + * + * @author Gary Russell + * @since 2.1.3 + * + */ +public class ChainedKafkaTransactionManager extends ChainedTransactionManager implements KafkaAwareTransactionManager { + + private final KafkaAwareTransactionManager kafkaTransactionManager; + + @SuppressWarnings("unchecked") + public ChainedKafkaTransactionManager(PlatformTransactionManager... transactionManagers) { + super(transactionManagers); + KafkaAwareTransactionManager kafkaTransactionManager = null; + for (PlatformTransactionManager tm : transactionManagers) { + if (tm instanceof KafkaAwareTransactionManager) { + Assert.isNull(kafkaTransactionManager, "Only one KafkaAwareTransactionManager is allowed"); + kafkaTransactionManager = (KafkaTransactionManager) tm; + } + } + Assert.notNull(kafkaTransactionManager, "Exactly one KafkaAwareTransactionManager is required"); + this.kafkaTransactionManager = kafkaTransactionManager; + } + + @Override + public ProducerFactory getProducerFactory() { + return this.kafkaTransactionManager.getProducerFactory(); + } + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/transaction/KafkaAwareTransactionManager.java b/spring-kafka/src/main/java/org/springframework/kafka/transaction/KafkaAwareTransactionManager.java new file mode 100644 index 00000000..be51bb44 --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/transaction/KafkaAwareTransactionManager.java @@ -0,0 +1,39 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.transaction; + +import org.springframework.kafka.core.ProducerFactory; + +/** + * A transaction manager that can provide a {@link ProducerFactory}. + * + * @param the key type. + * @param the value type. + * + * @author Gary Russell + * @since 2.1.3 + * + */ +public interface KafkaAwareTransactionManager { + + /** + * Get the producer factory. + * @return the producerFactory + */ + ProducerFactory getProducerFactory(); + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/transaction/KafkaTransactionManager.java b/spring-kafka/src/main/java/org/springframework/kafka/transaction/KafkaTransactionManager.java index a2f5fc28..64e1c983 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/transaction/KafkaTransactionManager.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/transaction/KafkaTransactionManager.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -66,7 +66,7 @@ import org.springframework.util.Assert; */ @SuppressWarnings("serial") public class KafkaTransactionManager extends AbstractPlatformTransactionManager - implements ResourceTransactionManager { + implements ResourceTransactionManager, KafkaAwareTransactionManager { private final ProducerFactory producerFactory; @@ -88,6 +88,7 @@ public class KafkaTransactionManager extends AbstractPlatformTransactionMa * Get the producer factory. * @return the producerFactory */ + @Override public ProducerFactory getProducerFactory() { return this.producerFactory; } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java index d5ca10f4..d1581314 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -64,7 +64,9 @@ import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.listener.config.ContainerProperties; import org.springframework.kafka.test.rule.KafkaEmbedded; import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.kafka.transaction.ChainedKafkaTransactionManager; import org.springframework.kafka.transaction.KafkaTransactionManager; +import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.TransactionException; import org.springframework.transaction.support.AbstractPlatformTransactionManager; @@ -88,9 +90,18 @@ public class TransactionalContainerTests { @ClassRule public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(3, true, topic1, topic2); - @SuppressWarnings({ "rawtypes", "unchecked" }) @Test - public void testConsumeAndProduceTransaction() throws Exception { + public void testConsumeAndProduceTransactionKTM() throws Exception { + testConsumeAndProduceTransactionGuts(false); + } + + @Test + public void testConsumeAndProduceTransactionKCTM() throws Exception { + testConsumeAndProduceTransactionGuts(true); + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + private void testConsumeAndProduceTransactionGuts(boolean chained) throws Exception { Consumer consumer = mock(Consumer.class); final TopicPartition topicPartition = new TopicPartition("foo", 0); willAnswer(i -> { @@ -122,9 +133,13 @@ public class TransactionalContainerTests { given(pf.transactionCapable()).willReturn(true); given(pf.createProducer()).willReturn(producer); KafkaTransactionManager tm = new KafkaTransactionManager(pf); + PlatformTransactionManager ptm = tm; + if (chained) { + ptm = new ChainedKafkaTransactionManager(new SomeOtherTransactionManager(), tm); + } ContainerProperties props = new ContainerProperties("foo"); props.setGroupId("group"); - props.setTransactionManager(tm); + props.setTransactionManager(ptm); final KafkaTemplate template = new KafkaTemplate(pf); props.setMessageListener((MessageListener) m -> { template.send("bar", "baz"); diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 8681956d..d54d31e0 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -280,7 +280,7 @@ If the listener throws an exception, the transaction is rolled back and the cons If you need to synchronize a Kafka transaction with some other transaction; simply 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 will participate in a single transaction. The Kafka transaction will be committed (or rolled back) immediately after the controlling transaction. -Before exiting the listener, you should invoke one of the template's `sendOffsetsToTransaction` methods. +Before exiting the listener, you should invoke one of the template's `sendOffsetsToTransaction` methods (unless you use a <>). For convenience, the listener container binds its consumer group id to the thread so, generally, you can use the first method: [source, java] @@ -317,6 +317,15 @@ NOTE: The offset to be committed is one greater than the offset of the record(s) IMPORTANT: This should only be called when using transaction synchronization. When a listener container is configured to use a `KafkaTransactionManager`, it will take care of sending the offsets to the transaction. +[[chained-transaction-manager]] +====== 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. +Chain your transaction managers in the desired order and provide the `ChainedTransactionManager` in the `ContainerProperties`. + ====== KafkaTemplate Local Transactions You can use the `KafkaTemplate` to execute a series of operations within a local transaction. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 86542c5e..51a9e352 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -42,6 +42,10 @@ See <> for more information. Starting with _version 2.1.3_, a subclass of `KafkaTemplate` is provided to support request/reply semantics. See <> for more information. +==== ChainedKafkaTransactionManager + +_version 2.1.3_ introduced the `ChainedKafkaTransactionManager` see <> 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].