Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Support ProducerRecord outbound payloads

Browse Source
This commit is contained in:
Gary Russell
2019-10-30 15:29:19 -04:00
committed by Artem Bilan
parent cf3aa29a23
commit 386c64c617
2 changed files with 88 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

102
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java
View File

@@ -59,6 +59,7 @@ import org.springframework.kafka.support.SimpleKafkaHeaderMapper;
import org.springframework.kafka.support.converter.KafkaMessageHeaders;
import org.springframework.kafka.support.converter.MessagingMessageConverter;
import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandlingException;
@@ -72,9 +73,17 @@ import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.util.concurrent.SettableListenableFuture;
/**
* Kafka Message Handler; when supplied with a {@link ReplyingKafkaTemplate}
* it is used as the handler in an outbound gateway. When supplied with a simple
* {@link KafkaTemplate} it used as the handler in an outbound channel adapter.
* Kafka Message Handler; when supplied with a {@link ReplyingKafkaTemplate} it is used as
* the handler in an outbound gateway. When supplied with a simple {@link KafkaTemplate}
* it used as the handler in an outbound channel adapter.
* <p>
* Starting with version 3.2.1 the handler supports receiving a pre-built
* {@link ProducerRecord} payload. In that case, most configuration properties
* ({@link #setTopicExpression(Expression)} etc.) are ignored. If the handler is used as
* gateway, the {@link ProducerRecord} will have its headers enhanced to add the
* {@link KafkaHeaders#REPLY_TOPIC} unless it already contains such a header. The handler
* will not map any additional headers; providing such a payload assumes the headers have
* already been mapped.
*
* @param <K> the key type.
* @param <V> the value type.
@@ -137,6 +146,8 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
(message, topic, partition, timestamp, key, value, headers) ->
new ProducerRecord<>(topic, partition, timestamp, key, value, headers);
private volatile byte[] singleReplyTopic;
public KafkaProducerMessageHandler(final KafkaTemplate<K, V> kafkaTemplate) {
Assert.notNull(kafkaTemplate, "kafkaTemplate cannot be null");
this.kafkaTemplate = kafkaTemplate;
@@ -373,6 +384,47 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
@SuppressWarnings("unchecked")
@Override
protected Object handleRequestMessage(final Message<?> message) {
final ProducerRecord<K, V> producerRecord;
boolean preBuilt = message.getPayload() instanceof ProducerRecord;
if (preBuilt) {
producerRecord = (ProducerRecord<K, V>) message.getPayload();
}
else {
producerRecord = createProducerRecord(message);
}
ListenableFuture<SendResult<K, V>> sendFuture;
RequestReplyFuture<K, V, Object> gatewayFuture = null;
if (this.isGateway && (!preBuilt || producerRecord.headers().lastHeader(KafkaHeaders.REPLY_TOPIC) == null)) {
producerRecord.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, getReplyTopic(message)));
gatewayFuture = ((ReplyingKafkaTemplate<K, V, Object>) this.kafkaTemplate).sendAndReceive(producerRecord);
sendFuture = gatewayFuture.getSendFuture();
}
else {
if (this.transactional
&& TransactionSynchronizationManager.getResource(this.kafkaTemplate.getProducerFactory()) == null) {
sendFuture = this.kafkaTemplate.executeInTransaction(t -> {
return t.send(producerRecord);
});
}
else {
sendFuture = this.kafkaTemplate.send(producerRecord);
}
}
try {
processSendResult(message, producerRecord, sendFuture, getSendSuccessChannel());
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new MessageHandlingException(message, e);
}
catch (ExecutionException e) {
throw new MessageHandlingException(message, e.getCause());
}
return processReplyFuture(gatewayFuture);
}
@SuppressWarnings("unchecked")
private ProducerRecord<K, V> createProducerRecord(final Message<?> message) {
MessageHeaders messageHeaders = message.getHeaders();
String topic = this.topicExpression != null ?
this.topicExpression.getValue(this.evaluationContext, message, String.class)
@@ -407,35 +459,7 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
}
final ProducerRecord<K, V> producerRecord = this.producerRecordCreator.create(message, topic, partitionId,
timestamp, (K) messageKey, payload, headers);
ListenableFuture<SendResult<K, V>> sendFuture;
RequestReplyFuture<K, V, Object> gatewayFuture = null;
if (this.isGateway) {
producerRecord.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, getReplyTopic(message)));
gatewayFuture = ((ReplyingKafkaTemplate<K, V, Object>) this.kafkaTemplate).sendAndReceive(producerRecord);
sendFuture = gatewayFuture.getSendFuture();
}
else {
if (this.transactional
&& TransactionSynchronizationManager.getResource(this.kafkaTemplate.getProducerFactory()) == null) {
sendFuture = this.kafkaTemplate.executeInTransaction(t -> {
return t.send(producerRecord);
});
}
else {
sendFuture = this.kafkaTemplate.send(producerRecord);
}
}
try {
processSendResult(message, producerRecord, sendFuture, getSendSuccessChannel());
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new MessageHandlingException(message, e);
}
catch (ExecutionException e) {
throw new MessageHandlingException(message, e.getCause());
}
return processReplyFuture(gatewayFuture);
return producerRecord;
}
private byte[] getReplyTopic(final Message<?> message) {
@@ -457,7 +481,7 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
}
if (replyTopic == null) {
if (this.replyTopicsAndPartitions.size() == 1) {
replyTopic = this.replyTopicsAndPartitions.keySet().iterator().next().getBytes(StandardCharsets.UTF_8);
replyTopic = getSingleReplyTopic();
}
else {
throw new IllegalStateException("No reply topic header and no default reply topic is can be determined");
@@ -487,6 +511,16 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
return replyTopic;
}
private byte[] getSingleReplyTopic() {
if (this.singleReplyTopic == null) {
this.singleReplyTopic = this.replyTopicsAndPartitions.keySet()
.iterator()
.next()
.getBytes(StandardCharsets.UTF_8);
}
return this.singleReplyTopic;
}
private void determineValidReplyTopicsAndPartitions() {
ReplyingKafkaTemplate<?, ?, ?> rkt = (ReplyingKafkaTemplate<?, ?, ?>) kafkaTemplate;
Collection<TopicPartition> replyTopics = rkt.getAssignedReplyTopicPartitions();
@@ -544,7 +578,7 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
}
}
private Future<?> processReplyFuture(RequestReplyFuture<?, ?, Object> future) {
private Future<?> processReplyFuture(@Nullable RequestReplyFuture<?, ?, Object> future) {
if (future == null) {
return null;
}

25
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java
View File

@@ -352,6 +352,15 @@ class KafkaProducerMessageHandlerTests {
@Test
void testOutboundGateway() throws Exception {
testOutboundGatewayGuts(null);
}
@Test
void testOutboundGatewayPrPayload() throws Exception {
testOutboundGatewayGuts(new ProducerRecord<Integer, String>(topic5, 1, 2, "foo"));
}
private void testOutboundGatewayGuts(ProducerRecord<?, ?> payload) throws Exception {
ConsumerFactory<Integer, String> consumerFactory = new DefaultKafkaConsumerFactory<>(
KafkaTestUtils.consumerProps(topic5, "false", embeddedKafka));
ContainerProperties containerProperties = new ContainerProperties(topic6);
@@ -384,11 +393,17 @@ class KafkaProducerMessageHandlerTests {
handler.setOutputChannel(replies);
handler.afterPropertiesSet();
Message<?> message = MessageBuilder.withPayload("foo")
.setHeader(KafkaHeaders.TOPIC, topic5)
.setHeader(KafkaHeaders.MESSAGE_KEY, 2)
.setHeader(KafkaHeaders.PARTITION_ID, 1)
.build();
Message<?> message;
if (payload == null) {
message = MessageBuilder.withPayload("foo")
.setHeader(KafkaHeaders.TOPIC, topic5)
.setHeader(KafkaHeaders.MESSAGE_KEY, 2)
.setHeader(KafkaHeaders.PARTITION_ID, 1)
.build();
}
else {
message = MessageBuilder.withPayload(payload).build();
}
handler.handleMessage(message);
ConsumerRecord<Integer, String> record = KafkaTestUtils.getSingleRecord(consumer, topic5);