测试容器中的 Kafka 侦听器，避免端口 9092

Question

我写了一个集成测试来证明发送给Kafka的消息将与侦听器一起到达。 它通过当且仅当我使用KAFKA_PORT=9092. 该常量是开发人员计算机(或 CI 计算机(上使用的端口。

最终，我想在动态分配的端口上执行此操作(即使用GenericContainer而不是FixedHostPortGenericContainer(，但目前我只想能够使用不同的端口。

如果我在下面的代码中设置 KAFKA_PORT=59092，那么测试就会失败，我看到控制台输出，例如Connection to node -1 (localhost/127.0.0.1:9092) could not be established：

2020-06-08 12:16:22.374  WARN 1371 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-1, groupId=test-consumer-group] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.

我假设我需要做一些额外的配置，所以没有 尝试使用端口 9092，但这让我无法接受。

下面是一个重新创建的剥离测试，以及相关的gradle.build。

KafaSpikeFixedPort.java

package com.example.kafkaspike;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.testcontainers.containers.FixedHostPortGenericContainer;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.assertAll;
import static org.junit.jupiter.api.Assertions.assertEquals;
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE)
@Testcontainers
public class KafaSpikeFixedPort {
// Test works only if this port is 9092 (matching the Docker container port)
final static int KAFKA_PORT = 9092;
@DynamicPropertySource
static void kafkaProperties(DynamicPropertyRegistry registry) {
registry.add("spring.kafka.producer.bootstrap-servers",
() -> "kubernetes.docker.internal:"+ KAFKA_PORT);
registry.add("spring.kafka.consumer.group-id",
() -> "test-consumer-group");
}
@Container
private GenericContainer kafkaContainer =
new FixedHostPortGenericContainer("obsidiandynamics/kafka:2.3.0-11")
.withFixedExposedPort(KAFKA_PORT, 9092)
.withExtraHost("kubernetes.docker.internal", "127.0.0.1")
.withEnv("KAFKA_LISTENERS",
"INTERNAL://:29092," +
"EXTERNAL://:"+KAFKA_PORT)
.withEnv("KAFKA_ADVERTISED_LISTENERS",
"INTERNAL://kubernetes.docker.internal:29092," +
"EXTERNAL://kubernetes.docker.internal:"+KAFKA_PORT)
.withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP",
"INTERNAL:PLAINTEXT," +
"EXTERNAL:PLAINTEXT")
.withEnv("KAFKA_INTER_BROKER_LISTENER_NAME",
"INTERNAL")
.waitingFor(Wait.forLogMessage(
".*INFO\s+\[KafkaServer\s+id=\d+\]" +
"\s+started\s+\(kafka.server.KafkaServer\).*",
1));
@Autowired
KafkaTemplate<String, String> kafkaTemplate;
@Autowired
KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;
private List<String> payloadsReceived = new ArrayList<>();
@KafkaListener(autoStartup = "false", topics = "topic1")
public void onMessage(@Payload String payload) {
payloadsReceived.add(payload);
}
@BeforeEach
public void beforeEach() {
payloadsReceived.clear();
for(MessageListenerContainer listenerContainer : kafkaListenerEndpointRegistry.getListenerContainers()) {
listenerContainer.start();
}
sleep(2_000); // Just for the spike. (Eliminates code checking the listener container states.)
}
@AfterEach
public void afterEach() {
for(MessageListenerContainer listenerContainer : kafkaListenerEndpointRegistry.getListenerContainers()) {
listenerContainer.stop();
}
sleep(2_000); // Just for the spike. (Eliminates code checking the listener container states.)
}
@Test
@Timeout(value = 3, unit = TimeUnit.SECONDS)
public void test() {
kafkaTemplate.send("topic1", "Hello World!");
sleep(1_000); // Just for the spike. (Wait for message in production test, with test timeout.)
assertAll(
() -> assertEquals(1, payloadsReceived.size()),
() -> assertEquals("Hello World!", payloadsReceived.get(0))
);
}
private void sleep(long millis) {
try {
Thread.sleep(millis);
} catch(InterruptedException e) {
}
}
}

gradle.build

plugins {
id 'org.springframework.boot' version '2.2.7.RELEASE'
id 'io.spring.dependency-management' version '1.0.9.RELEASE'
id 'java'
}
group = 'com.example'
version = '0.0.1-SNAPSHOT'
sourceCompatibility = '11'
configurations {
compileOnly {
extendsFrom annotationProcessor
}
}
repositories {
mavenCentral()
}
dependencies {
implementation 'org.springframework.boot:spring-boot-starter'
implementation 'org.springframework.kafka:spring-kafka'
compileOnly 'org.projectlombok:lombok'
annotationProcessor 'org.springframework.boot:spring-boot-configuration-processor'
annotationProcessor 'org.projectlombok:lombok'
testImplementation('org.springframework.boot:spring-boot-starter-test') {
exclude group: 'org.junit.vintage', module: 'junit-vintage-engine'
}
testImplementation 'org.springframework.kafka:spring-kafka-test'
// for TestContainers
testCompile group: 'org.testcontainers', name: 'testcontainers', version: "1.14.1"
testCompile group: 'org.testcontainers', name: 'junit-jupiter', version: "1.14.1"
}
test {
useJUnitPlatform()
}

Answer 1

Testcontainers的Kafka模块存在是有原因的:)

它通过延迟运行 Kafka 的进程来处理端口设置，以便它可以提供实际分配的随机端口作为 env 变量ADVERTISED_HOST。

尝试一下，或者看看来源以获得一些灵感：
https://github.com/testcontainers/testcontainers-java/tree/master/modules/kafka