From 905a5155bc7497d4d6a84d863e7a28888d55df44 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 20 Feb 2020 12:30:39 -0500 Subject: [PATCH] Master to 3.3; support non-Tx if template allows * SI to 5.3.0.B-S --- .../outbound/KafkaProducerMessageHandler.java | 14 +++++++++---- .../xml/KafkaOutboundAdapterParserTests.java | 5 ++--- .../KafkaProducerMessageHandlerTests.java | 20 ++++++++++++++++--- 3 files changed, 29 insertions(+), 10 deletions(-) diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java index f7a6cd41a2..f53f31ed00 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java @@ -110,6 +110,8 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes private final boolean transactional; + private final boolean allowNonTransactional; + private final AtomicBoolean running = new AtomicBoolean(); private EvaluationContext evaluationContext; @@ -164,6 +166,7 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes this.headerMapper = new SimpleKafkaHeaderMapper(); } this.transactional = kafkaTemplate.isTransactional(); + this.allowNonTransactional = kafkaTemplate.isAllowNonTransactional(); if (this.transactional && this.isGateway) { logger.warn("The KafkaTemplate is transactional; this gateway will only work if the consumer is " + "configured to read uncommitted records"); @@ -372,7 +375,9 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes @Override public void stop() { if (this.running.compareAndSet(true, false)) { - this.kafkaTemplate.flush(); + if (!this.transactional || this.allowNonTransactional) { + this.kafkaTemplate.flush(); + } } } @@ -401,9 +406,10 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes } else { if (this.transactional - && TransactionSynchronizationManager.getResource(this.kafkaTemplate.getProducerFactory()) == null) { - sendFuture = this.kafkaTemplate.executeInTransaction(t -> { - return t.send(producerRecord); + && TransactionSynchronizationManager.getResource(this.kafkaTemplate.getProducerFactory()) == null + && !this.allowNonTransactional) { + sendFuture = this.kafkaTemplate.executeInTransaction(template -> { + return template.send(producerRecord); }); } else { diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java index 00bc903d20..997cceaddd 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2019 the original author or authors. + * Copyright 2013-2020 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. @@ -18,7 +18,6 @@ package org.springframework.integration.kafka.config.xml; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatExceptionOfType; -import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.BDDMockito.given; import static org.mockito.Mockito.mock; @@ -107,7 +106,7 @@ class KafkaOutboundAdapterParserTests { }; @SuppressWarnings("unchecked") ProducerFactory pf = mock(ProducerFactory.class); - given(pf.createProducer(isNull())).willReturn(mockProducer); + given(pf.createProducer()).willReturn(mockProducer); KafkaTemplate template = new KafkaTemplate<>(pf); KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler<>(template); handler.setBeanFactory(mock(BeanFactory.class)); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java index d94932c6ec..533b1b6a0c 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-2020 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. @@ -25,6 +25,7 @@ import static org.mockito.BDDMockito.willAnswer; import static org.mockito.BDDMockito.willReturn; import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoMoreInteractions; @@ -474,7 +475,6 @@ class KafkaProducerMessageHandlerTests { inOrder.verify(producer).beginTransaction(); inOrder.verify(producer).send(any(ProducerRecord.class), any(Callback.class)); inOrder.verify(producer).commitTransaction(); - inOrder.verify(producer).flush(); } @SuppressWarnings({ "unchecked", "rawtypes" }) @@ -584,7 +584,6 @@ class KafkaProducerMessageHandlerTests { inOrder.verify(producer).beginTransaction(); inOrder.verify(producer).send(any(ProducerRecord.class), any(Callback.class)); inOrder.verify(producer).commitTransaction(); - inOrder.verify(producer).flush(); assertThat(txId.get()).isEqualTo("overridden.tx.id."); } @@ -674,4 +673,19 @@ class KafkaProducerMessageHandlerTests { assertThat(txId.get()).isEqualTo("tm.tx.id."); } + @SuppressWarnings("unchecked") + @Test + void testTxNonTx() { + KafkaTemplate template = mock(KafkaTemplate.class); + given(template.isTransactional()).willReturn(true); + given(template.inTransaction()).willReturn(false); + given(template.isAllowNonTransactional()).willReturn(true); + given(template.getProducerFactory()).willReturn(mock(ProducerFactory.class)); + KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler<>(template); + handler.setTopicExpression(new LiteralExpression("topic")); + handler.handleMessage(new GenericMessage<>("foo")); + verify(template, never()).executeInTransaction(any()); + verify(template).send(any(ProducerRecord.class)); + } + }