4343import org .springframework .kafka .core .ProducerFactory ;
4444import org .springframework .kafka .event .ContainerStoppedEvent ;
4545import org .springframework .kafka .test .EmbeddedKafkaBroker ;
46+ import org .springframework .kafka .test .condition .LogLevels ;
4647import org .springframework .kafka .test .context .EmbeddedKafka ;
4748import org .springframework .kafka .test .utils .KafkaTestUtils ;
4849import org .springframework .test .context .junit .jupiter .SpringJUnitConfig ;
@@ -59,25 +60,32 @@ public class ContainerGroupSequencerTests {
5960 private static final LogAccessor LOGGER = new LogAccessor (LogFactory .getLog (ContainerGroupSequencerTests .class ));
6061
6162 @ Test
62- void sequenceCompletes (@ Autowired Config config , @ Autowired KafkaTemplate <Integer , String > template )
63+ @ LogLevels (classes = { ContainerGroupSequencerTests .class , ContainerGroupSequencer .class }, level = "DEBUG" )
64+ void sequenceCompletes (@ Autowired Config config , @ Autowired KafkaTemplate <Integer , String > template ,
65+ @ Autowired ContainerGroupSequencer sequencer )
6366 throws InterruptedException {
6467
68+ sequencer .start ();
6569 template .send ("ContainerGroupSequencerTests" , "test" );
6670 assertThat (config .stopped .await (10 , TimeUnit .SECONDS ))
6771 .as ("stopped latch still has a count of %d" , config .stopped .getCount ())
6872 .isTrue ();
6973 List <String > order = config .order ;
70- assertThat (order .get (0 ))
71- .as ("out of order %s" )
74+ String expected = order .get (0 );
75+ assertThat (expected )
76+ .as ("out of order %s" , expected )
7277 .isIn ("one" , "two" );
73- assertThat (order .get (1 ))
74- .as ("out of order %s" )
78+ expected = order .get (1 );
79+ assertThat (expected )
80+ .as ("out of order %s" , expected )
7581 .isIn ("one" , "two" );
76- assertThat (order .get (2 ))
77- .as ("out of order %s" )
82+ expected = order .get (2 );
83+ assertThat (expected )
84+ .as ("out of order %s" , expected )
7885 .isIn ("three" , "four" );
79- assertThat (order .get (3 ))
80- .as ("out of order %s" )
86+ expected = order .get (3 );
87+ assertThat (expected )
88+ .as ("out of order %s" , expected )
8189 .isIn ("three" , "four" );
8290 assertThat (config .receivedAt .get (3 ) - config .receivedAt .get (0 )).isGreaterThanOrEqualTo (1000 );
8391 }
@@ -95,24 +103,28 @@ public static class Config {
95103
96104 @ KafkaListener (id = "one" , topics = "ContainerGroupSequencerTests" , containerGroup = "g1" , concurrency = "2" )
97105 public void listen1 (String in ) {
106+ LOGGER .debug (in );
98107 this .order .add ("one" );
99108 this .receivedAt .add (System .currentTimeMillis ());
100109 }
101110
102111 @ KafkaListener (id = "two" , topics = "ContainerGroupSequencerTests" , containerGroup = "g1" , concurrency = "2" )
103112 public void listen2 (String in ) {
113+ LOGGER .debug (in );
104114 this .order .add ("two" );
105115 this .receivedAt .add (System .currentTimeMillis ());
106116 }
107117
108118 @ KafkaListener (id = "three" , topics = "ContainerGroupSequencerTests" , containerGroup = "g2" , concurrency = "2" )
109119 public void listen3 (String in ) {
120+ LOGGER .debug (in );
110121 this .order .add ("three" );
111122 this .receivedAt .add (System .currentTimeMillis ());
112123 }
113124
114125 @ KafkaListener (id = "four" , topics = "ContainerGroupSequencerTests" , containerGroup = "g2" , concurrency = "2" )
115126 public void listen4 (String in ) {
127+ LOGGER .debug (in );
116128 this .order .add ("four" );
117129 this .receivedAt .add (System .currentTimeMillis ());
118130 }
@@ -127,6 +139,7 @@ public void stopped(ContainerStoppedEvent event) {
127139 ContainerGroupSequencer sequencer (KafkaListenerEndpointRegistry registry ) {
128140 ContainerGroupSequencer sequencer = new ContainerGroupSequencer (registry , 1000 , "g1" , "g2" );
129141 sequencer .setStopLastGroupWhenIdle (true );
142+ sequencer .setAutoStartup (false );
130143 return sequencer ;
131144 }
132145
0 commit comments