From 39dd68a8ada1315e1148694e7ba65bd18a6fca3b Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Sun, 31 Dec 2017 14:01:31 -0500 Subject: [PATCH] SCST-GH-1166: Support Producer-initiated tx Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/1166 Previously, transactions were only supported if initiated by a consumer. Also fix races in `KafkaBinderTests.testResume()` (messages sent before subscribing). --- .../kafka/KafkaMessageChannelBinder.java | 30 +++++- .../stream/binder/kafka/KafkaBinderTests.java | 19 ++-- .../binder/kafka/KafkaTransactionTests.java | 97 +++++++++++++++++++ 3 files changed, 136 insertions(+), 10 deletions(-) create mode 100644 spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTransactionTests.java 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 8a0638f22..eddc886e9 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 @@ -66,6 +66,7 @@ import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAd import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy; import org.springframework.integration.support.ErrorMessageStrategy; +import org.springframework.kafka.KafkaException; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; @@ -82,9 +83,12 @@ import org.springframework.kafka.support.SendResult; import org.springframework.kafka.support.TopicPartitionInitialOffset; import org.springframework.kafka.support.converter.MessagingMessageConverter; import org.springframework.kafka.transaction.KafkaTransactionManager; +import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHeaders; +import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.transaction.support.TransactionTemplate; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; @@ -122,6 +126,8 @@ public class KafkaMessageChannelBinder extends private final KafkaTransactionManager transactionManager; + private final TransactionTemplate transactionTemplate; + private ProducerListener producerListener; private KafkaExtendedBindingProperties extendedBindingProperties = new KafkaExtendedBindingProperties(); @@ -134,9 +140,11 @@ public class KafkaMessageChannelBinder extends this.transactionManager = new KafkaTransactionManager<>( getProducerFactory(configurationProperties.getTransaction().getTransactionIdPrefix(), new ExtendedProducerProperties<>(configurationProperties.getTransaction().getProducer()))); + this.transactionTemplate = new TransactionTemplate(this.transactionManager); } else { this.transactionManager = null; + this.transactionTemplate = null; } } @@ -260,7 +268,7 @@ public class KafkaMessageChannelBinder extends return handler; } - private DefaultKafkaProducerFactory getProducerFactory(String transactionIdPrefix, + protected DefaultKafkaProducerFactory getProducerFactory(String transactionIdPrefix, ExtendedProducerProperties producerProperties) { Map props = new HashMap<>(); props.put(ProducerConfig.RETRIES_CONFIG, 0); @@ -611,6 +619,26 @@ public class KafkaMessageChannelBinder extends public boolean isRunning() { return this.running; } + + @Override + protected void handleMessageInternal(Message message) throws Exception { + if (KafkaMessageChannelBinder.this.transactionTemplate != null + && !TransactionSynchronizationManager.isActualTransactionActive()) { + KafkaMessageChannelBinder.this.transactionTemplate.execute(s -> { + try { + super.handleMessageInternal(message); + } + catch (Exception e) { + throw new KafkaException("Exception on transactional send", e); + } + return null; + }); + } + else { + super.handleMessageInternal(message); + } + } + } static class TopicInformation { diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 45d68a310..92f047b1e 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -33,6 +33,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import com.fasterxml.jackson.databind.ObjectMapper; + import org.apache.kafka.clients.admin.AdminClient; import org.apache.kafka.clients.admin.AdminClientConfig; import org.apache.kafka.clients.admin.CreateTopicsResult; @@ -1815,8 +1816,6 @@ public class KafkaBinderTests extends String testTopicName = UUID.randomUUID().toString(); producerBinding = binder.bindProducer(testTopicName, output, producerBindingProperties.getProducer()); - String testPayload1 = "foo1-" + UUID.randomUUID().toString(); - output.send(new GenericMessage<>(testPayload1)); ExtendedConsumerProperties firstConsumerProperties = createConsumerProperties(); consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, firstConsumerProperties); @@ -1831,12 +1830,12 @@ public class KafkaBinderTests extends } }; input1.subscribe(messageHandler); - Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message"); + String testPayload1 = "foo1-" + UUID.randomUUID().toString(); + output.send(new GenericMessage<>(testPayload1)); + Assert.isTrue(latch.await(15, TimeUnit.SECONDS), "Failed to receive message"); assertThat(inboundMessageRef1.get()).isNotNull(); assertThat(inboundMessageRef1.get().getPayload()).isNotNull(); - String testPayload2 = "foo2-" + UUID.randomUUID().toString(); - output.send(new GenericMessage<>(testPayload2.getBytes())); input1.unsubscribe(messageHandler); CountDownLatch latch1 = new CountDownLatch(1); AtomicReference> inboundMessageRef2 = new AtomicReference<>(); @@ -1849,14 +1848,14 @@ public class KafkaBinderTests extends } }; input1.subscribe(messageHandler1); - Assert.isTrue(latch1.await(5, TimeUnit.SECONDS), "Failed to receive message"); + String testPayload2 = "foo2-" + UUID.randomUUID().toString(); + output.send(new GenericMessage<>(testPayload2.getBytes())); + Assert.isTrue(latch1.await(15, TimeUnit.SECONDS), "Failed to receive message"); assertThat(inboundMessageRef2.get()).isNotNull(); assertThat(inboundMessageRef2.get().getPayload()).isNotNull(); consumerBinding.unbind(); Thread.sleep(2000); - String testPayload3 = "foo3-" + UUID.randomUUID().toString(); - output.send(new GenericMessage<>(testPayload3.getBytes())); ExtendedConsumerProperties consumerProperties = createConsumerProperties(); consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, consumerProperties); input1.unsubscribe(messageHandler1); @@ -1871,7 +1870,9 @@ public class KafkaBinderTests extends } }; input1.subscribe(messageHandler2); - Assert.isTrue(latch2.await(5, TimeUnit.SECONDS), "Failed to receive message"); + String testPayload3 = "foo3-" + UUID.randomUUID().toString(); + output.send(new GenericMessage<>(testPayload3.getBytes())); + Assert.isTrue(latch2.await(15, TimeUnit.SECONDS), "Failed to receive message"); assertThat(inboundMessageRef3.get()).isNotNull(); assertThat(new String(inboundMessageRef3.get().getPayload())).isEqualTo(testPayload3); } diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTransactionTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTransactionTests.java new file mode 100644 index 000000000..8b81ee4e9 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTransactionTests.java @@ -0,0 +1,97 @@ +/* + * 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.cloud.stream.binder.kafka; + +import java.util.Collections; + +import org.apache.kafka.clients.producer.Callback; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.TopicPartition; +import org.junit.ClassRule; +import org.junit.Test; +import org.mockito.InOrder; + +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties; +import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; +import org.springframework.context.support.GenericApplicationContext; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.test.rule.KafkaEmbedded; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.retry.support.RetryTemplate; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.BDDMockito.willReturn; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; + +/** + * @author Gary Russell + * @since 2.0 + * + */ +public class KafkaTransactionTests { + + @ClassRule + public static final KafkaEmbedded embeddedKafka = new KafkaEmbedded(1); + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Test + public void testProducerRunsInTx() { + KafkaProperties kafkaProperties = new KafkaProperties(); + kafkaProperties.setBootstrapServers(Collections.singletonList(embeddedKafka.getBrokersAsString())); + KafkaBinderConfigurationProperties configurationProperties = new KafkaBinderConfigurationProperties(); + configurationProperties.getTransaction().setTransactionIdPrefix("foo-"); + KafkaTopicProvisioner provisioningProvider = new KafkaTopicProvisioner(configurationProperties, kafkaProperties); + provisioningProvider.setMetadataRetryOperations(new RetryTemplate()); + final Producer mockProducer = mock(Producer.class); + willReturn(Collections.singletonList(new TopicPartition("foo", 0))).given(mockProducer).partitionsFor(anyString()); + KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(configurationProperties, provisioningProvider) { + + @Override + protected DefaultKafkaProducerFactory getProducerFactory(String transactionIdPrefix, + ExtendedProducerProperties producerProperties) { + DefaultKafkaProducerFactory producerFactory = + spy(super.getProducerFactory(transactionIdPrefix, producerProperties)); + willReturn(mockProducer).given(producerFactory).createProducer(); + return producerFactory; + } + + }; + GenericApplicationContext applicationContext = new GenericApplicationContext(); + applicationContext.refresh(); + binder.setApplicationContext(applicationContext); + DirectChannel channel = new DirectChannel(); + KafkaProducerProperties extension = new KafkaProducerProperties(); + ExtendedProducerProperties properties = new ExtendedProducerProperties<>(extension); + binder.bindProducer("foo", channel, properties); + channel.send(new GenericMessage<>("foo".getBytes())); + InOrder inOrder = inOrder(mockProducer); + inOrder.verify(mockProducer).beginTransaction(); + inOrder.verify(mockProducer).send(any(ProducerRecord.class), any(Callback.class)); + inOrder.verify(mockProducer).commitTransaction(); + inOrder.verify(mockProducer).close(); + inOrder.verifyNoMoreInteractions(); + } + +}