diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index 5828247c7..9306bce11 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/docs/src/main/asciidoc/overview.adoc @@ -506,6 +506,50 @@ public class Application { } ---- +[[kafka-transactional-binder]] +=== Transactional Binder + +Enable transactions by setting `spring.cloud.stream.kafka.binder.transaction.transactionIdPrefix` to a non-empty value, e.g. `tx-`. +When used in a processor application, the consumer starts the transaction; any records sent on the consumer thread participate in the same transaction. +When the listener exits normally, the listener container will send the offset to the transaction and commit it. +A common producer factory is used for all producer bindings configure using `spring.cloud.stream.kafka.binder.transaction.producer.*` properties; individual binding Kafka producer properties are ignored. + +If you wish to use transactions in a source application, or from some arbitrary thread for producer-only transaction (e.g. `@Scheduled` method), you must get a reference to the transactional producer factory and define a `KafkaTransactionManager` bean using it. + +==== +[source, java] +---- +@Bean +public PlatformTransactionManager transactionManager(BinderFactory binders) { + ProducerFactory pf = ((KafkaMessageChannelBinder) binders.getBinder(null, + MessageChannel.class)).getTransactionalProducerFactory(); + return new KafkaTransactionManager<>(pf); +} +---- +==== + +Notice that we get a reference to the binder using the `BinderFactory`; use `null` in the first argument when there is only one binder configured. +If more than one binder is configured, use the binder name to get the reference. +Once we have a reference to the binder, we can obtain a reference to the `ProducerFactory` and create a transaction manager. + +Then you would just normal Spring transaction support, e.g. `TransactionTemplate` or `@Transactional`, for example: + +==== +[source, java] +---- +public static class Sender { + + @Transactional + public void doInTransaction(MessageChannel output, List stuffToSend) { + stuffToSend.forEach(stuff -> output.send(new GenericMessage<>(stuff))); + } + +} +---- +==== + +If you wish to synchronize producer-only transactions with those from some other transaction manager, use a `ChainedTransactionManager`. + [[kafka-error-channels]] === Error Channels diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 4a6608ec4..940f77673 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -105,6 +105,7 @@ import org.springframework.kafka.support.TopicPartitionInitialOffset; import org.springframework.kafka.support.TopicPartitionInitialOffset.SeekPosition; import org.springframework.kafka.support.converter.MessagingMessageConverter; import org.springframework.kafka.transaction.KafkaTransactionManager; +import org.springframework.lang.Nullable; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHeaders; @@ -287,6 +288,17 @@ public class KafkaMessageChannelBinder extends return this.extendedBindingProperties.getExtendedPropertiesEntryClass(); } + /** + * Return a reference to the binder's transaction manager's producer factory (if + * configured). Use this to create a transaction manager in a bean definition when you + * wish to use producer-only transactions. + * @return the transaction manager, or null. + */ + @Nullable + public ProducerFactory getTransactionalProducerFactory() { + return this.transactionManager == null ? null : this.transactionManager.getProducerFactory(); + } + @Override protected MessageHandler createProducerMessageHandler( final ProducerDestination destination, diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/ProducerOnlyTransactionTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/ProducerOnlyTransactionTests.java new file mode 100644 index 000000000..d43af0486 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/ProducerOnlyTransactionTests.java @@ -0,0 +1,152 @@ +/* + * Copyright 2019-2019 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 + * + * https://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.cloud.stream.binder.kafka.integration; + +import java.util.Map; + +import kafka.server.KafkaConfig; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.common.requests.IsolationLevel; +import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.BeanCreationException; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.binder.BinderFactory; +import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder; +import org.springframework.cloud.stream.messaging.Source; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.kafka.transaction.KafkaTransactionManager; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.annotation.EnableTransactionManagement; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.AbstractPlatformTransactionManager; +import org.springframework.transaction.support.TransactionSynchronizationManager; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Gary Russell + * @since 2.1.4 + * + */ +@RunWith(SpringRunner.class) +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = { + "spring.cloud.stream.kafka.binder.transaction.transaction-id-prefix=tx.", + "spring.cloud.stream.kafka.binder.transaction.producer.configuration.retries=99", + "spring.cloud.stream.kafka.binder.transaction.producer.configuration.acks=all"}) +public class ProducerOnlyTransactionTests { + + private static final String KAFKA_BROKERS_PROPERTY = "spring.cloud.stream.kafka.binder.brokers"; + + @ClassRule + public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, "output") + .brokerProperty(KafkaConfig.TransactionsTopicReplicationFactorProp(), "1") + .brokerProperty(KafkaConfig.TransactionsTopicMinISRProp(), "1"); + + @Autowired + private Sender sender; + + @Autowired + private MessageChannel output; + + @BeforeClass + public static void setup() { + System.setProperty(KAFKA_BROKERS_PROPERTY, + embeddedKafka.getEmbeddedKafka().getBrokersAsString()); + } + + @AfterClass + public static void clean() { + System.clearProperty(KAFKA_BROKERS_PROPERTY); + } + + @Test + public void testProducerTx() { + this.sender.DoInTransaction(this.output); + assertThat(this.sender.isInTx()).isTrue(); + Map props = KafkaTestUtils.consumerProps("consumeTx", "false", + embeddedKafka.getEmbeddedKafka()); + props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); + props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_COMMITTED.name().toLowerCase()); + Consumer consumer = new KafkaConsumer<>(props); + embeddedKafka.getEmbeddedKafka().consumeFromAllEmbeddedTopics(consumer); + ConsumerRecord record = KafkaTestUtils.getSingleRecord(consumer, "output"); + assertThat(record.value()).isEqualTo("foo".getBytes()); + } + + @EnableBinding(Source.class) + @EnableAutoConfiguration + @EnableTransactionManagement + public static class Config { + + @Bean + public PlatformTransactionManager transactionManager(BinderFactory binders) { + try { + ProducerFactory pf = ((KafkaMessageChannelBinder) binders.getBinder(null, + MessageChannel.class)).getTransactionalProducerFactory(); + KafkaTransactionManager tm = new KafkaTransactionManager<>(pf); + tm.setTransactionSynchronization(AbstractPlatformTransactionManager.SYNCHRONIZATION_ON_ACTUAL_TRANSACTION); + return tm; + } + catch (BeanCreationException e) { // needed to avoid other tests in this package failing when there is no binder + return null; + } + } + + @Bean + public Sender sender() { + return new Sender(); + } + + } + + public static class Sender { + + private boolean isInTx; + + @Transactional + public void DoInTransaction(MessageChannel output) { + this.isInTx = TransactionSynchronizationManager.isActualTransactionActive(); + output.send(new GenericMessage<>("foo")); + } + + public boolean isInTx() { + return this.isInTx; + } + + } + +}