SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

SCST-GH-1166: Support Producer-initiated tx

Browse Source 
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).
This commit is contained in:
Gary Russell
2017-12-31 14:01:31 -05:00
committed by Oleg Zhurakousky
parent 8daaed43a9
commit 39dd68a8ad
3 changed files with 136 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -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<byte[], byte[]> transactionManager;
private final TransactionTemplate transactionTemplate;
private ProducerListener<byte[], byte[]> 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<byte[], byte[]> getProducerFactory(String transactionIdPrefix,
protected DefaultKafkaProducerFactory<byte[], byte[]> getProducerFactory(String transactionIdPrefix,
ExtendedProducerProperties<KafkaProducerProperties> producerProperties) {
Map<String, Object> 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 {

19
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
View File

@@ -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<KafkaConsumerProperties> 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<Message<String>> 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<KafkaConsumerProperties> 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);
}

97
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTransactionTests.java Normal file
View File

@@ -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<byte[], byte[]> getProducerFactory(String transactionIdPrefix,
ExtendedProducerProperties<KafkaProducerProperties> producerProperties) {
DefaultKafkaProducerFactory<byte[], byte[]> 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<KafkaProducerProperties> 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();
}
}