Skip to content

Commit ac39bc2

Browse files
committed
unit test listener transaction nested kafka transaction
* kafka listener nested kafka transactions with @transactional does not support retryable topic
1 parent 049aab9 commit ac39bc2

File tree

1 file changed

+67
-9
lines changed

1 file changed

+67
-9
lines changed

spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicTransactionIntegrationTests.java

Lines changed: 67 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,7 @@
5454
import org.springframework.stereotype.Component;
5555
import org.springframework.test.annotation.DirtiesContext;
5656
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
57+
import org.springframework.transaction.annotation.Transactional;
5758

5859
/**
5960
* {@link org.springframework.kafka.annotation.RetryableTopic} with transaction.
@@ -63,27 +64,33 @@
6364
@SpringJUnitConfig
6465
@DirtiesContext
6566
@EmbeddedKafka(topics = { RetryTopicTransactionIntegrationTests.RETRY_TRANSACTION_FIRST_TOPIC,
66-
RetryTopicTransactionIntegrationTests.RETRY_KAFKA_LISTENER_NESTED_TX_FIRST_TOPIC }, partitions = 1,
67+
RetryTopicTransactionIntegrationTests.RETRY_KAFKA_LISTENER_NESTED_TX_FIRST_TOPIC,
68+
RetryTopicTransactionIntegrationTests.RETRY_TRANSACTION_SECOND_TOPIC,
69+
RetryTopicTransactionIntegrationTests.RETRY_KAFKA_LISTENER_NESTED_TX_SECOND_TOPIC}, partitions = 1,
6770
brokerProperties = { "transaction.state.log.replication.factor=1", "transaction.state.log.min.isr=1" })
6871
public class RetryTopicTransactionIntegrationTests {
6972

7073
public final static String RETRY_TRANSACTION_FIRST_TOPIC = "retryTopic1";
7174

7275
public final static String RETRY_KAFKA_LISTENER_NESTED_TX_FIRST_TOPIC = "retryNestedTxTopic1";
7376

77+
public final static String RETRY_TRANSACTION_SECOND_TOPIC = "retryTopic2";
78+
79+
public final static String RETRY_KAFKA_LISTENER_NESTED_TX_SECOND_TOPIC = "retryNestedTxTopic2";
80+
7481
@Autowired
7582
private KafkaTemplate<String, String> kafkaTemplate;
7683

7784
@Autowired
7885
private CountDownLatchContainer latchContainer;
7986

8087
@Test
81-
@DisplayName("retry topic not support kafka listener nested kafka transactions")
88+
@DisplayName("kafka listener nested kafka transactions does not support retryable topic")
8289
void shouldRetryableTopicWithKafkaListenerNestedKafkaTransactions() {
8390
kafkaTemplate.executeInTransaction(t ->
8491
kafkaTemplate.send(RETRY_TRANSACTION_FIRST_TOPIC, "Testing topic 1")
8592
);
86-
assertThat(awaitLatch(latchContainer.countDownLatchFirstRetryable)).isTrue();
93+
assertThat(awaitLatch(latchContainer.countDownLatchOneRetryable)).isTrue();
8794
assertThat(awaitLatch(latchContainer.countDownLatchDltOne)).isTrue();
8895
ConsumerRecord<String, String> consumerRecord =
8996
kafkaTemplate.receive(RETRY_KAFKA_LISTENER_NESTED_TX_FIRST_TOPIC, 0, 4);
@@ -92,9 +99,24 @@ void shouldRetryableTopicWithKafkaListenerNestedKafkaTransactions() {
9299
assertThat(consumerRecord.offset()).isEqualTo(4);
93100
}
94101

102+
@Test
103+
@DisplayName("kafka listener nested kafka transactions with @Transactional does not support retryable topic")
104+
void shouldRetryableTopicWithKafkaListenerNestedKafkaTransactionsAndTransactional() {
105+
kafkaTemplate.executeInTransaction(t ->
106+
kafkaTemplate.send(RETRY_TRANSACTION_SECOND_TOPIC, "Testing topic 2")
107+
);
108+
assertThat(awaitLatch(latchContainer.countDownLatchTwoRetryable)).isTrue();
109+
assertThat(awaitLatch(latchContainer.countDownLatchDltTwo)).isTrue();
110+
ConsumerRecord<String, String> consumerRecord =
111+
kafkaTemplate.receive(RETRY_KAFKA_LISTENER_NESTED_TX_SECOND_TOPIC, 0, 4);
112+
assertThat(consumerRecord).isNotNull();
113+
assertThat(consumerRecord.value()).isEqualTo("message-3");
114+
assertThat(consumerRecord.offset()).isEqualTo(4);
115+
}
116+
95117
private boolean awaitLatch(CountDownLatch latch) {
96118
try {
97-
return latch.await(150, TimeUnit.SECONDS);
119+
return latch.await(10, TimeUnit.SECONDS);
98120
}
99121
catch (Exception e) {
100122
fail(e.getMessage());
@@ -103,7 +125,7 @@ private boolean awaitLatch(CountDownLatch latch) {
103125
}
104126

105127
@Component
106-
static class FirstRetryableKafkaListener {
128+
static class KafkaListenerWithRetryableAndNoTransactionalAnnotation {
107129

108130
@Autowired
109131
CountDownLatchContainer countDownLatchContainer;
@@ -116,7 +138,7 @@ static class FirstRetryableKafkaListener {
116138
@RetryableTopic(topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE)
117139
@KafkaListener(topics = RetryTopicTransactionIntegrationTests.RETRY_TRANSACTION_FIRST_TOPIC)
118140
void listen(String in, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
119-
countDownLatchContainer.countDownLatchFirstRetryable.countDown();
141+
countDownLatchContainer.countDownLatchOneRetryable.countDown();
120142
kafkaTemplate.send(RETRY_KAFKA_LISTENER_NESTED_TX_FIRST_TOPIC, "m-" + ++SEND_MESSAGE_COUNT);
121143
throw new RuntimeException("from FirstRetryableKafkaListener");
122144
}
@@ -125,14 +147,45 @@ void listen(String in, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
125147
void dlt(String in, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
126148
countDownLatchContainer.countDownLatchDltOne.countDown();
127149
}
150+
151+
}
152+
153+
@Component
154+
static class KafkaListenerWithRetryableAndTransactionalAnnotation {
155+
156+
@Autowired
157+
CountDownLatchContainer countDownLatchContainer;
158+
159+
@Autowired
160+
KafkaTemplate<String, String> kafkaTemplate;
161+
162+
static int SEND_MESSAGE_COUNT = 0;
163+
164+
@RetryableTopic(topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE)
165+
@KafkaListener(topics = RetryTopicTransactionIntegrationTests.RETRY_TRANSACTION_SECOND_TOPIC)
166+
@Transactional
167+
void listen(String in, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
168+
countDownLatchContainer.countDownLatchTwoRetryable.countDown();
169+
kafkaTemplate.send(RETRY_KAFKA_LISTENER_NESTED_TX_SECOND_TOPIC, "message-" + ++SEND_MESSAGE_COUNT);
170+
throw new RuntimeException("from SecondRetryableKafkaListener");
171+
}
172+
173+
@DltHandler
174+
void dlt(String in, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
175+
countDownLatchContainer.countDownLatchDltTwo.countDown();
176+
}
177+
128178
}
129179

130180
@Component
131181
static class CountDownLatchContainer {
132182

133-
CountDownLatch countDownLatchFirstRetryable = new CountDownLatch(3);
183+
CountDownLatch countDownLatchOneRetryable = new CountDownLatch(3);
134184
CountDownLatch countDownLatchDltOne = new CountDownLatch(1);
135185

186+
CountDownLatch countDownLatchTwoRetryable = new CountDownLatch(3);
187+
CountDownLatch countDownLatchDltTwo = new CountDownLatch(1);
188+
136189
}
137190

138191
@EnableKafka
@@ -148,8 +201,13 @@ CountDownLatchContainer latchContainer() {
148201
}
149202

150203
@Bean
151-
FirstRetryableKafkaListener firstRetryableKafkaListener() {
152-
return new FirstRetryableKafkaListener();
204+
KafkaListenerWithRetryableAndNoTransactionalAnnotation kafkaListenerWithRetryableAndNoTx() {
205+
return new KafkaListenerWithRetryableAndNoTransactionalAnnotation();
206+
}
207+
208+
@Bean
209+
KafkaListenerWithRetryableAndTransactionalAnnotation kafkaListenerWithRetryableAndTx() {
210+
return new KafkaListenerWithRetryableAndTransactionalAnnotation();
153211
}
154212

155213
@Bean

0 commit comments

Comments
 (0)