Sending message from transactional Kafka listener: Invalid transition attempted from state IN_TRANSACTION to state IN_TRANSACTION #645
Comments
eugene-khyst
changed the title
|
Your test kafkaTemplate.executeInTransaction(kt ->
kt.send(INPUT_TEST_TOPIC, testKey, testData));(or don't use a transactional template). I also fixed your asserts: assertEquals(testKey, record.key());
assertEquals(testData, record.value());(they were both comparing the value). |
|
@garyrussell thank you. I've checked when test |
|
You would need to define two producer factories and two templates (one with and one without a transaction id prefix). I just issued a PR #647 to detect the condition and provide a meaningful exception. |
|
@garyrussell, did the test pass for you? I tried to run the test and assertion fails:
It seems that listener method doesn't consume the event. I removed the transaction manager from the listener container and annotated |
|
Yes, it works for me - I just ran it again...
package com.example.kafka.sample.transaction;
import static com.example.kafka.sample.transaction.listener.TestKafkaListener.INPUT_TEST_TOPIC;
import static com.example.kafka.sample.transaction.listener.TestKafkaListener.OUTPUT_TEST_TOPIC;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import java.util.Iterator;
import java.util.Map;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.test.rule.KafkaEmbedded;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.test.context.TestPropertySource;
import org.springframework.test.context.junit4.SpringRunner;
@TestPropertySource("classpath:test.properties")
@RunWith(SpringRunner.class)
@SpringBootTest
public class KafkaListenerTest {
@Autowired
private KafkaEmbedded kafkaEmbedded;
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
@Test
public void shouldProcessEvent() throws Exception {
String testKey = "test_key";
String testData = "test_data";
kafkaTemplate.executeInTransaction(kt ->
kt.send(INPUT_TEST_TOPIC, testKey, testData));
try (Consumer<String, String> consumer = createConsumer()) {
kafkaEmbedded.consumeFromAnEmbeddedTopic(consumer, OUTPUT_TEST_TOPIC);
ConsumerRecords<String, String> records = KafkaTestUtils.getRecords(consumer);
Iterator<ConsumerRecord<String, String>> iterator = records.iterator();
ConsumerRecord<String, String> record = iterator.next();
assertEquals(testKey, record.key());
assertEquals(testData, record.value());
assertFalse(iterator.hasNext());
}
}
private Consumer<String, String> createConsumer() {
Map<String, Object> consumerProps =
KafkaTestUtils.consumerProps("test-consumer", "true", kafkaEmbedded);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(
consumerProps, new StringDeserializer(), new StringDeserializer());
return cf.createConsumer();
}
} |
|
Test fails on Windows but passes on Ubuntu... probably the issue is related to Windows |
Probably the issue is related to the Apache Kafka which doesn't support Windows properly. Yes, I develop on Windows as well and I really see some issues with Kafka very often, especially with transactions. I think the issue comes out of the scope of this project. You can try to raise issue against Apache Kafka and cross-link it here. Thanks for understanding |
|
@eklierka this is a known Kafka issue: https://issues.apache.org/jira/browse/KAFKA-6052. According to the ticket, fix will be included in the Kafka version 1.1.1. |
|
Closing for now; we can reopen if this tuns out to be a spring-kafka issue. |
|
Hello @eklierka and all! I tried exactly the above in Windows (spring-kafka-2.1.6.RELEASE and kafka 1.0.1) as well and the behavior I am seeing is that the TestKafkaListener.listen() receives the record successfully, but OUTPUT_TEST_TOPIC never received the record. Is this the issue you had on Windows? From my logs: |
|
Yes, I'm observing some issue with Kafka transaction on Windows. Thanks |
|
Thanks @artembilan, It was indeed a windows related issue, as in my mac works fine! |
|
Hi @vspiliopoulos . |
I'm trying to send a message from Kafka listener in the same transaction to have message sending and offset committing in the same transaction but receives an exception: Invalid transition attempted from state IN_TRANSACTION to state IN_TRANSACTION, and the transaction is rolled back.
I've added transaction support to my application (Spring Boot 2, Spring Kafka 2.1.4) using the following approach.
application.properties
I've set kafkaTransactionManager to kafkaListenerContainerFactory containerProperties:
Test listener should receive a message from input topic (transaction-sample-topic-in), log it, and forward it to the output topic (transaction-sample-topic-out). I want message sending and offset committing to be performed in the same transaction.
In the tests, I've defined embedded Kafka with broker properties allowing to have transaction support with 1 broker instance.
When I run a simple test that sends a message to the input topic I receive an exception: Invalid transition attempted from state IN_TRANSACTION to state IN_TRANSACTION. The transaction is rolled back and the message is not forwarded to the output topic.
The exception is
I've created a sample application to reproduce the problem: https://github.com/evgeniy-khist/spring-kafka-transactions-sample
The text was updated successfully, but these errors were encountered: