garyrussell
commented
7 years ago What version are you testing with? Can you get a DEBUG log? I just ran this test and everything worked as expected...
@Test
public void testDeclarativeIntegration() throws Exception {
AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(DeclarativeConfig.class,
EmbeddedConfig.class);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testTxString", "false", embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
cf.setKeyDeserializer(new StringDeserializer());
Consumer<String, String> consumer = cf.createConsumer();
embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "fiz");
Tx2 tx2 = ctx.getBean(Tx2.class);
tx2.anotherTxMethod();
ConsumerRecord<String, String> record = KafkaTestUtils.getSingleRecord(consumer, "fiz");
assertThat(record.value()).isEqualTo("buz");
tx2.anotherTxMethod();
record = KafkaTestUtils.getSingleRecord(consumer, "fiz");
assertThat(record.value()).isEqualTo("buz");
ctx.close();
}
public static class Tx2 {
@SuppressWarnings("rawtypes")
private final KafkaTemplate template;
@SuppressWarnings("rawtypes")
public Tx2(KafkaTemplate template) {
this.template = template;
}
@SuppressWarnings("unchecked")
@Transactional
public void anotherTxMethod() {
template.send("fiz", "buz");
}
}11:28:56.422 DEBUG [main][org.springframework.kafka.transaction.KafkaTransactionManager] Creating new transaction with name [org.springframework.kafka.core.KafkaTemplateTransactionTests$Tx2.anotherTxMethod]: PROPAGATION_REQUIRED,ISOLATION_DEFAULT; ''
11:28:58.936 DEBUG [main][org.springframework.kafka.transaction.KafkaTransactionManager] Created Kafka transaction on producer [CloseSafeProducer [delegate=org.apache.kafka.clients.producer.KafkaProducer@327c7bea]]
11:28:59.184 DEBUG [main][org.springframework.kafka.transaction.KafkaTransactionManager] Initiating transaction commit
11:28:59.350 DEBUG [main][org.springframework.kafka.transaction.KafkaTransactionManager] Creating new transaction with name [org.springframework.kafka.core.KafkaTemplateTransactionTests$Tx2.anotherTxMethod]: PROPAGATION_REQUIRED,ISOLATION_DEFAULT; ''
11:28:59.350 DEBUG [main][org.springframework.kafka.transaction.KafkaTransactionManager] Created Kafka transaction on producer [CloseSafeProducer [delegate=org.apache.kafka.clients.producer.KafkaProducer@327c7bea]]
11:28:59.350 DEBUG [main][org.springframework.kafka.transaction.KafkaTransactionManager] Initiating transaction commit
agrawal93
tomaszszlek
artembilan
hajlaoui-nader
msfkos
To add transaction support in my application, I followed the below mentioned approach:
@Configuration @EnableKafka @EnableTransactionManagement public class KafkaProducerConfig {
}
After calling the
publishAsTransactionmethod with appropriatetopicandmessages[], the first run is successful with all messages getting published to the topic. But, every subsequent run after that gives me the errorInvalid transition attempted from state IN_TRANSACTION to state IN_TRANSACTION.Is there anything wrong with my approach? Kindly guide me on the same.