Skip to content

Commit 65b518f

Browse files
smcvbsmcvb
committed
Replace RedPanda for KafkaContainer
Replace RedPanda for KafkaContainer. The new version of Test containers no longer likes the red panda version we pulled in. Also fix warnings for the touched files. #552
1 parent a773544 commit 65b518f

File tree

3 files changed

+34
-31
lines changed

3 files changed

+34
-31
lines changed

‎kafka-spring-boot-3-integrationtests/pom.xml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -82,7 +82,7 @@
8282
</dependency>
8383
<dependency>
8484
<groupId>org.testcontainers</groupId>
85-
<artifactId>redpanda</artifactId>
85+
<artifactId>kafka</artifactId>
8686
<scope>test</scope>
8787
</dependency>
8888
<dependency>

‎kafka-spring-boot-3-integrationtests/src/test/java/org/axonframework/extensions/kafka/integration/StreamableKafkaSourceIntegrationTest.java

Lines changed: 14 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@
2929
import org.springframework.test.context.ContextConfiguration;
3030
import org.testcontainers.junit.jupiter.Container;
3131
import org.testcontainers.junit.jupiter.Testcontainers;
32-
import org.testcontainers.redpanda.RedpandaContainer;
32+
import org.testcontainers.kafka.KafkaContainer;
3333

3434
import java.time.Duration;
3535

@@ -41,8 +41,8 @@
4141
class StreamableKafkaSourceIntegrationTest {
4242

4343
@Container
44-
private static final RedpandaContainer REDPANDA_CONTAINER = new RedpandaContainer(
45-
"docker.redpanda.com/vectorized/redpanda:v22.2.1");
44+
private static final KafkaContainer KAFKA_CONTAINER = new KafkaContainer("apache/kafka-native");
45+
4646
private ApplicationContextRunner testApplicationContext;
4747

4848
@BeforeEach
@@ -56,15 +56,18 @@ void messageSendViaKafkaShouldBeReceived() {
5656
.withPropertyValues("axon.axonserver.enabled=false")
5757
.withPropertyValues("axon.kafka.fetcher.enabled=true")
5858
.withPropertyValues("axon.kafka.consumer.event-processor-mode=tracking")
59-
.withPropertyValues("axon.kafka.producer.bootstrap-servers=" + REDPANDA_CONTAINER.getBootstrapServers())
60-
.withPropertyValues("axon.kafka.consumer.bootstrap-servers=" + REDPANDA_CONTAINER.getBootstrapServers())
59+
.withPropertyValues("axon.kafka.producer.bootstrap-servers=" + KAFKA_CONTAINER.getBootstrapServers())
60+
.withPropertyValues("axon.kafka.consumer.bootstrap-servers=" + KAFKA_CONTAINER.getBootstrapServers())
6161
.withUserConfiguration(DefaultContext.class)
6262
.run(context -> {
6363
EventGateway eventGateway = context.getBean(EventGateway.class);
6464
assertNotNull(eventGateway);
6565
publishEvent(eventGateway);
66-
StreamableKafkaMessageSource<String, byte[]> messageSource = context.getBean(
67-
StreamableKafkaMessageSource.class);
66+
67+
//noinspection unchecked
68+
StreamableKafkaMessageSource<String, byte[]> messageSource =
69+
context.getBean(StreamableKafkaMessageSource.class);
70+
6871
assertNotNull(messageSource);
6972
receiveMessage(messageSource);
7073
});
@@ -75,8 +78,10 @@ private void publishEvent(EventGateway eventGateway) {
7578
eventGateway.publish(event);
7679
}
7780

78-
private void receiveMessage(StreamableKafkaMessageSource<String, byte[]> messageSource)
79-
throws InterruptedException {
81+
private void receiveMessage(
82+
StreamableKafkaMessageSource<String, byte[]> messageSource
83+
) throws InterruptedException {
84+
//noinspection resource
8085
BlockingStream<TrackedEventMessage<?>> stream = messageSource.openStream(null);
8186
await().atMost(Duration.ofSeconds(5L)).until(stream::hasNextAvailable);
8287
TrackedEventMessage<?> message = stream.nextAvailable();

‎kafka-spring-boot-3-integrationtests/src/test/java/org/axonframework/extensions/kafka/integration/TokenReplayIntegrationTest.java

Lines changed: 19 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@
3939
import org.springframework.test.context.ContextConfiguration;
4040
import org.testcontainers.junit.jupiter.Container;
4141
import org.testcontainers.junit.jupiter.Testcontainers;
42-
import org.testcontainers.redpanda.RedpandaContainer;
42+
import org.testcontainers.kafka.KafkaContainer;
4343

4444
import java.net.URI;
4545
import java.time.Duration;
@@ -56,10 +56,9 @@
5656
class 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");
6260

61+
private ApplicationContextRunner testApplicationContext;
6362

6463
@BeforeEach
6564
void setUp() {
@@ -69,27 +68,27 @@ void setUp() {
6968
.withPropertyValues("axon.kafka.publisher.enabled=false")
7069
.withPropertyValues("axon.kafka.message-converter-mode=cloud_event")
7170
.withPropertyValues("axon.kafka.consumer.event-processor-mode=tracking")
72-
.withPropertyValues("axon.kafka.consumer.bootstrap-servers=" + REDPANDA_CONTAINER.getBootstrapServers())
71+
.withPropertyValues("axon.kafka.consumer.bootstrap-servers=" + KAFKA_CONTAINER.getBootstrapServers())
7372
.withUserConfiguration(DefaultContext.class);
7473
}
7574

7675
@Test
7776
void afterResetShouldOnlyProcessTenEventsIfTimeSetMidway() {
7877
testApplicationContext
79-
.withPropertyValues("axon.kafka.default-topic=counterfeed-1")
78+
.withPropertyValues("axon.kafka.default-topic=counter-feed-1")
8079
.run(context -> {
8180
Counter counter = context.getBean(Counter.class);
8281
assertNotNull(counter);
8382
assertEquals(0, counter.getCount());
84-
Instant between = addRecords("counterfeed-1");
83+
Instant between = addRecords("counter-feed-1");
8584
await().atMost(Duration.ofSeconds(5L)).untilAsserted(
8685
() -> assertEquals(20, counter.getCount())
8786
);
8887
EventProcessingConfiguration processingConfiguration = context.getBean(EventProcessingConfiguration.class);
8988
assertNotNull(processingConfiguration);
9089
processingConfiguration
9190
.eventProcessorByProcessingGroup(
92-
"counterfeedprocessor",
91+
"counter-feed-processor",
9392
TrackingEventProcessor.class
9493
)
9594
.ifPresent(tep -> {
@@ -107,20 +106,20 @@ void afterResetShouldOnlyProcessTenEventsIfTimeSetMidway() {
107106
@Test
108107
void afterResetShouldOnlyProcessNewMessages() {
109108
testApplicationContext
110-
.withPropertyValues("axon.kafka.default-topic=counterfeed-2")
109+
.withPropertyValues("axon.kafka.default-topic=counter-feed-2")
111110
.run(context -> {
112111
Counter counter = context.getBean(Counter.class);
113112
assertNotNull(counter);
114113
assertEquals(0, counter.getCount());
115-
addRecords("counterfeed-2");
114+
addRecords("counter-feed-2");
116115
await().atMost(Duration.ofSeconds(5L)).untilAsserted(
117116
() -> assertEquals(20, counter.getCount())
118117
);
119118
EventProcessingConfiguration processingConfiguration = context.getBean(EventProcessingConfiguration.class);
120119
assertNotNull(processingConfiguration);
121120
processingConfiguration
122121
.eventProcessorByProcessingGroup(
123-
"counterfeedprocessor",
122+
"counter-feed-processor",
124123
TrackingEventProcessor.class
125124
)
126125
.ifPresent(tep -> {
@@ -129,15 +128,15 @@ void afterResetShouldOnlyProcessNewMessages() {
129128
assertEquals(0, counter.getCount());
130129
tep.start();
131130
});
132-
addRecords("counterfeed-2");
131+
addRecords("counter-feed-2");
133132
await().atMost(Duration.ofSeconds(5L)).untilAsserted(
134133
() -> assertEquals(20, counter.getCount())
135134
);
136135
});
137136
}
138137

139138
private Instant addRecords(String topic) {
140-
Producer<String, CloudEvent> producer = newProducer(REDPANDA_CONTAINER.getBootstrapServers());
139+
Producer<String, CloudEvent> producer = newProducer(KAFKA_CONTAINER.getBootstrapServers());
141140
sendTenMessages(producer, topic);
142141
Instant now = Instant.now();
143142
sendTenMessages(producer, topic);
@@ -146,12 +145,11 @@ private Instant addRecords(String topic) {
146145
}
147146

148147
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();
148+
CloudEvent event = new CloudEventBuilder().withId(UUID.randomUUID().toString())
149+
.withSource(URI.create("source"))
150+
.withData("Payload".getBytes())
151+
.withType("java.util.String")
152+
.build();
155153
ProducerRecord<String, CloudEvent> record = new ProducerRecord<>(topic, 0, null, null, event);
156154
producer.send(record);
157155
}
@@ -182,7 +180,7 @@ public void registerProcessor(
182180
StreamableKafkaMessageSource<?, ?> streamableKafkaMessageSource
183181
) {
184182
configurer.eventProcessing()
185-
.registerTrackingEventProcessor("counterfeedprocessor", c -> streamableKafkaMessageSource);
183+
.registerTrackingEventProcessor("counter-feed-processor", c -> streamableKafkaMessageSource);
186184
}
187185
}
188186

@@ -205,7 +203,7 @@ void reset() {
205203

206204
@SuppressWarnings("unused")
207205
@Component
208-
@ProcessingGroup("counterfeedprocessor")
206+
@ProcessingGroup("counter-feed-processor")
209207
private static class KafkaEventHandler {
210208

211209
private final Counter counter;

0 commit comments

Comments
 (0)