11/*
2- * Copyright 2017 the original author or authors.
2+ * Copyright 2017-2018 the original author or authors.
33 *
44 * Licensed under the Apache License, Version 2.0 (the "License");
55 * you may not use this file except in compliance with the License.
2020
2121import java .util .Collections ;
2222import java .util .Map ;
23+ import java .util .concurrent .BlockingQueue ;
24+ import java .util .concurrent .CountDownLatch ;
25+ import java .util .concurrent .TimeUnit ;
2326
2427import org .apache .kafka .clients .consumer .ConsumerConfig ;
2528import org .apache .kafka .clients .consumer .KafkaConsumer ;
29+ import org .apache .kafka .clients .producer .ProducerConfig ;
2630import org .junit .Test ;
31+ import org .junit .runner .RunWith ;
32+
33+ import org .springframework .beans .factory .annotation .Autowired ;
34+ import org .springframework .context .annotation .Configuration ;
35+ import org .springframework .kafka .listener .KafkaMessageListenerContainer ;
36+ import org .springframework .kafka .listener .MessageListener ;
37+ import org .springframework .kafka .listener .config .ContainerProperties ;
38+ import org .springframework .kafka .support .SendResult ;
39+ import org .springframework .kafka .test .context .EmbeddedKafka ;
40+ import org .springframework .kafka .test .rule .KafkaEmbedded ;
41+ import org .springframework .kafka .test .utils .KafkaTestUtils ;
42+ import org .springframework .kafka .transaction .KafkaTransactionManager ;
43+ import org .springframework .test .annotation .DirtiesContext ;
44+ import org .springframework .test .context .junit4 .SpringRunner ;
45+ import org .springframework .util .concurrent .ListenableFuture ;
2746
2847/**
2948 * @author Gary Russell
3049 * @since 1.0.6
3150 *
3251 */
52+ @ EmbeddedKafka (topics = { "txCache1" , "txCache2" , "txCacheSendFromListener" },
53+ brokerProperties = {
54+ "transaction.state.log.replication.factor=1" ,
55+ "transaction.state.log.min.isr=1" }
56+ )
57+ @ RunWith (SpringRunner .class )
58+ @ DirtiesContext
3359public class DefaultKafkaConsumerFactoryTests {
3460
61+ @ Autowired
62+ private KafkaEmbedded embeddedKafka ;
63+
3564 @ Test
3665 public void testClientId () {
3766 Map <String , Object > configs = Collections .singletonMap (ConsumerConfig .CLIENT_ID_CONFIG , "foo" );
38- DefaultKafkaConsumerFactory <String , String > factory = new DefaultKafkaConsumerFactory <String , String >(configs ) {
67+ DefaultKafkaConsumerFactory <String , String > factory =
68+ new DefaultKafkaConsumerFactory <String , String >(configs ) {
3969
4070 @ Override
4171 protected KafkaConsumer <String , String > createKafkaConsumer (Map <String , Object > configs ) {
@@ -47,4 +77,87 @@ protected KafkaConsumer<String, String> createKafkaConsumer(Map<String, Object>
4777 factory .createConsumer ("-1" );
4878 }
4979
80+ @ SuppressWarnings ("unchecked" )
81+ @ Test
82+ public void testNestedTxProducerIsCached () throws Exception {
83+ Map <String , Object > producerProps = KafkaTestUtils .producerProps (this .embeddedKafka );
84+ producerProps .put (ProducerConfig .RETRIES_CONFIG , 1 );
85+ DefaultKafkaProducerFactory <Integer , String > pf = new DefaultKafkaProducerFactory <>(producerProps );
86+ KafkaTemplate <Integer , String > template = new KafkaTemplate <>(pf );
87+ DefaultKafkaProducerFactory <Integer , String > pfTx = new DefaultKafkaProducerFactory <>(producerProps );
88+ pfTx .setTransactionIdPrefix ("fooTx." );
89+ KafkaTemplate <Integer , String > templateTx = new KafkaTemplate <>(pfTx );
90+ Map <String , Object > consumerProps = KafkaTestUtils .consumerProps ("txCache1Group" , "false" , this .embeddedKafka );
91+ consumerProps .put (ConsumerConfig .AUTO_OFFSET_RESET_CONFIG , "earliest" );
92+ DefaultKafkaConsumerFactory <Integer , String > cf = new DefaultKafkaConsumerFactory <>(consumerProps );
93+ ContainerProperties containerProps = new ContainerProperties ("txCache1" );
94+ CountDownLatch latch = new CountDownLatch (1 );
95+ containerProps .setMessageListener ((MessageListener <Integer , String >) r -> {
96+ templateTx .executeInTransaction (t -> t .send ("txCacheSendFromListener" , "bar" ));
97+ templateTx .executeInTransaction (t -> t .send ("txCacheSendFromListener" , "baz" ));
98+ latch .countDown ();
99+ });
100+ KafkaTransactionManager <Integer , String > tm = new KafkaTransactionManager <>(pfTx );
101+ containerProps .setTransactionManager (tm );
102+ KafkaMessageListenerContainer <Integer , String > container = new KafkaMessageListenerContainer <>(cf ,
103+ containerProps );
104+ container .start ();
105+ try {
106+ ListenableFuture <SendResult <Integer , String >> future = template .send ("txCache1" , "foo" );
107+ future .get ();
108+ assertThat (KafkaTestUtils .getPropertyValue (pf , "cache" , BlockingQueue .class )).hasSize (0 );
109+ assertThat (latch .await (30 , TimeUnit .SECONDS )).isTrue ();
110+ assertThat (KafkaTestUtils .getPropertyValue (pfTx , "cache" , BlockingQueue .class )).hasSize (1 );
111+ }
112+ finally {
113+ container .stop ();
114+ pf .destroy ();
115+ pfTx .destroy ();
116+ }
117+ }
118+
119+ @ SuppressWarnings ("unchecked" )
120+ @ Test
121+ public void testContainerTxProducerIsNotCached () throws Exception {
122+ Map <String , Object > producerProps = KafkaTestUtils .producerProps (this .embeddedKafka );
123+ producerProps .put (ProducerConfig .RETRIES_CONFIG , 1 );
124+ DefaultKafkaProducerFactory <Integer , String > pf = new DefaultKafkaProducerFactory <>(producerProps );
125+ KafkaTemplate <Integer , String > template = new KafkaTemplate <>(pf );
126+ DefaultKafkaProducerFactory <Integer , String > pfTx = new DefaultKafkaProducerFactory <>(producerProps );
127+ pfTx .setTransactionIdPrefix ("fooTx." );
128+ KafkaTemplate <Integer , String > templateTx = new KafkaTemplate <>(pfTx );
129+ Map <String , Object > consumerProps = KafkaTestUtils .consumerProps ("txCache2Group" , "false" , this .embeddedKafka );
130+ consumerProps .put (ConsumerConfig .AUTO_OFFSET_RESET_CONFIG , "earliest" );
131+ DefaultKafkaConsumerFactory <Integer , String > cf = new DefaultKafkaConsumerFactory <>(consumerProps );
132+ ContainerProperties containerProps = new ContainerProperties ("txCache2" );
133+ CountDownLatch latch = new CountDownLatch (1 );
134+ containerProps .setMessageListener ((MessageListener <Integer , String >) r -> {
135+ templateTx .send ("txCacheSendFromListener" , "bar" );
136+ templateTx .send ("txCacheSendFromListener" , "baz" );
137+ latch .countDown ();
138+ });
139+ KafkaTransactionManager <Integer , String > tm = new KafkaTransactionManager <>(pfTx );
140+ containerProps .setTransactionManager (tm );
141+ KafkaMessageListenerContainer <Integer , String > container = new KafkaMessageListenerContainer <>(cf ,
142+ containerProps );
143+ container .start ();
144+ try {
145+ ListenableFuture <SendResult <Integer , String >> future = template .send ("txCache2" , "foo" );
146+ future .get ();
147+ assertThat (KafkaTestUtils .getPropertyValue (pf , "cache" , BlockingQueue .class )).hasSize (0 );
148+ assertThat (latch .await (30 , TimeUnit .SECONDS )).isTrue ();
149+ assertThat (KafkaTestUtils .getPropertyValue (pfTx , "cache" , BlockingQueue .class )).hasSize (0 );
150+ }
151+ finally {
152+ container .stop ();
153+ pf .destroy ();
154+ pfTx .destroy ();
155+ }
156+ }
157+
158+ @ Configuration
159+ public static class Config {
160+
161+ }
162+
50163}
0 commit comments