3939import org .springframework .test .context .ContextConfiguration ;
4040import org .testcontainers .junit .jupiter .Container ;
4141import org .testcontainers .junit .jupiter .Testcontainers ;
42- import org .testcontainers .redpanda . RedpandaContainer ;
42+ import org .testcontainers .kafka . KafkaContainer ;
4343
4444import java .net .URI ;
4545import java .time .Duration ;
5656class TokenReplayIntegrationTest {
5757
5858 @ Container
59- private static final RedpandaContainer REDPANDA_CONTAINER = new RedpandaContainer (
60- "docker.redpanda.com/vectorized/redpanda:v22.2.1" );
61- private ApplicationContextRunner testApplicationContext ;
59+ private static final KafkaContainer KAFKA_CONTAINER = new KafkaContainer ("apache/kafka-native" )
60+ .withEnv ("KAFKA_LISTENERS" , "PLAINTEXT://:9092,BROKER://:9093,CONTROLLER://:9094" );
6261
62+ private ApplicationContextRunner testApplicationContext ;
6363
6464 @ BeforeEach
6565 void setUp () {
@@ -69,27 +69,27 @@ void setUp() {
6969 .withPropertyValues ("axon.kafka.publisher.enabled=false" )
7070 .withPropertyValues ("axon.kafka.message-converter-mode=cloud_event" )
7171 .withPropertyValues ("axon.kafka.consumer.event-processor-mode=tracking" )
72- .withPropertyValues ("axon.kafka.consumer.bootstrap-servers=" + REDPANDA_CONTAINER .getBootstrapServers ())
72+ .withPropertyValues ("axon.kafka.consumer.bootstrap-servers=" + KAFKA_CONTAINER .getBootstrapServers ())
7373 .withUserConfiguration (DefaultContext .class );
7474 }
7575
7676 @ Test
7777 void afterResetShouldOnlyProcessTenEventsIfTimeSetMidway () {
7878 testApplicationContext
79- .withPropertyValues ("axon.kafka.default-topic=counterfeed -1" )
79+ .withPropertyValues ("axon.kafka.default-topic=counter-feed -1" )
8080 .run (context -> {
8181 Counter counter = context .getBean (Counter .class );
8282 assertNotNull (counter );
8383 assertEquals (0 , counter .getCount ());
84- Instant between = addRecords ("counterfeed -1" );
84+ Instant between = addRecords ("counter-feed -1" );
8585 await ().atMost (Duration .ofSeconds (5L )).untilAsserted (
8686 () -> assertEquals (20 , counter .getCount ())
8787 );
8888 EventProcessingConfiguration processingConfiguration = context .getBean (EventProcessingConfiguration .class );
8989 assertNotNull (processingConfiguration );
9090 processingConfiguration
9191 .eventProcessorByProcessingGroup (
92- "counterfeedprocessor " ,
92+ "counter-feed-processor " ,
9393 TrackingEventProcessor .class
9494 )
9595 .ifPresent (tep -> {
@@ -107,20 +107,20 @@ void afterResetShouldOnlyProcessTenEventsIfTimeSetMidway() {
107107 @ Test
108108 void afterResetShouldOnlyProcessNewMessages () {
109109 testApplicationContext
110- .withPropertyValues ("axon.kafka.default-topic=counterfeed -2" )
110+ .withPropertyValues ("axon.kafka.default-topic=counter-feed -2" )
111111 .run (context -> {
112112 Counter counter = context .getBean (Counter .class );
113113 assertNotNull (counter );
114114 assertEquals (0 , counter .getCount ());
115- addRecords ("counterfeed -2" );
115+ addRecords ("counter-feed -2" );
116116 await ().atMost (Duration .ofSeconds (5L )).untilAsserted (
117117 () -> assertEquals (20 , counter .getCount ())
118118 );
119119 EventProcessingConfiguration processingConfiguration = context .getBean (EventProcessingConfiguration .class );
120120 assertNotNull (processingConfiguration );
121121 processingConfiguration
122122 .eventProcessorByProcessingGroup (
123- "counterfeedprocessor " ,
123+ "counter-feed-processor " ,
124124 TrackingEventProcessor .class
125125 )
126126 .ifPresent (tep -> {
@@ -129,15 +129,15 @@ void afterResetShouldOnlyProcessNewMessages() {
129129 assertEquals (0 , counter .getCount ());
130130 tep .start ();
131131 });
132- addRecords ("counterfeed -2" );
132+ addRecords ("counter-feed -2" );
133133 await ().atMost (Duration .ofSeconds (5L )).untilAsserted (
134134 () -> assertEquals (20 , counter .getCount ())
135135 );
136136 });
137137 }
138138
139139 private Instant addRecords (String topic ) {
140- Producer <String , CloudEvent > producer = newProducer (REDPANDA_CONTAINER .getBootstrapServers ());
140+ Producer <String , CloudEvent > producer = newProducer (KAFKA_CONTAINER .getBootstrapServers ());
141141 sendTenMessages (producer , topic );
142142 Instant now = Instant .now ();
143143 sendTenMessages (producer , topic );
@@ -146,12 +146,11 @@ private Instant addRecords(String topic) {
146146 }
147147
148148 private void sendMessage (Producer <String , CloudEvent > producer , String topic ) {
149- CloudEvent event = new CloudEventBuilder ()
150- .withId (UUID .randomUUID ().toString ())
151- .withSource (URI .create ("source" ))
152- .withData ("Payload" .getBytes ())
153- .withType ("java.util.String" )
154- .build ();
149+ CloudEvent event = new CloudEventBuilder ().withId (UUID .randomUUID ().toString ())
150+ .withSource (URI .create ("source" ))
151+ .withData ("Payload" .getBytes ())
152+ .withType ("java.util.String" )
153+ .build ();
155154 ProducerRecord <String , CloudEvent > record = new ProducerRecord <>(topic , 0 , null , null , event );
156155 producer .send (record );
157156 }
@@ -182,7 +181,7 @@ public void registerProcessor(
182181 StreamableKafkaMessageSource <?, ?> streamableKafkaMessageSource
183182 ) {
184183 configurer .eventProcessing ()
185- .registerTrackingEventProcessor ("counterfeedprocessor " , c -> streamableKafkaMessageSource );
184+ .registerTrackingEventProcessor ("counter-feed-processor " , c -> streamableKafkaMessageSource );
186185 }
187186 }
188187
@@ -205,7 +204,7 @@ void reset() {
205204
206205 @ SuppressWarnings ("unused" )
207206 @ Component
208- @ ProcessingGroup ("counterfeedprocessor " )
207+ @ ProcessingGroup ("counter-feed-processor " )
209208 private static class KafkaEventHandler {
210209
211210 private final Counter counter ;
0 commit comments