Spring Kafka测试-未在带有嵌入式Kafka的@KafkaListener中接收数据

Question

我们正在使用Cucumber为out应用程序做一些集成测试，我们在测试一个@KafkaListener时遇到了一些问题。我们设法使用了一个嵌入式Kafka并在其中生成数据。

但消费者从未收到任何数据，我们也不知道发生了什么。

这是我们的代码：

制作人配置

@Configuration
@Profile("test")
public class KafkaTestProducerConfig {

    private static final String SCHEMA_REGISTRY_URL = "schema.registry.url";

    @Autowired
    protected EmbeddedKafkaBroker embeddedKafka;

    @Bean
    public Map<String, Object> producerConfig() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                embeddedKafka.getBrokersAsString());
        props.put(SCHEMA_REGISTRY_URL, "URL");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
        return props;
    }

    @Bean
    public ProducerFactory<String, GenericRecord> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfig());
    }

    @Bean
    public KafkaTemplate<String, GenericRecord> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

}

消费者配置

@Configuration
@Profile("test")
@EnableKafka
public class KafkaTestConsumerConfig {

    @Autowired
    protected EmbeddedKafkaBroker embeddedKafka;

    private static final String SCHEMA_REGISTRY_URL = "schema.registry.url";

    @Bean
    public Map<String, Object> consumerProperties() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "groupId");
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());
        props.put(SCHEMA_REGISTRY_URL, "URL");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1000");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 10000);
        return props;
    }

    @Bean
    public DefaultKafkaConsumerFactory<String, Object> consumerFactory() {
        KafkaAvroDeserializer avroDeserializer = new KafkaAvroDeserializer();
        avroDeserializer.configure(consumerProperties(), false);
        return new DefaultKafkaConsumerFactory<>(consumerProperties(), new StringDeserializer(), avroDeserializer);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setBatchListener(true);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }

}

集成测试

@SpringBootTest(
        webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
        classes = Application.class)
@ActiveProfiles("test")
@EmbeddedKafka(topics = {"TOPIC1", "TOPIC2", "TOPIC3"})
public class CommonStepDefinitions implements En {

    protected static final Logger LOGGER = LoggerFactory.getLogger(CommonStepDefinitions.class);

    @Autowired
    protected KafkaTemplate<String, GenericRecord> kafkaTemplate;

}

步骤定义

public class KafkaStepDefinitions extends CommonStepDefinitions {

    private static final String TEMPLATE_TOPIC = "TOPIC1";

    public KafkaStepDefinitions(){
        Given("given statement", () -> {
            OperationEntity operationEntity = new OperationEntity();
            operationEntity.setFoo("foo");
            kafkaTemplate.send(TEMPLATE_TOPIC, AvroPojoTransformer.pojoToRecord(operationEntity));
        });
    }

}

消费者表示，同样的代码在生产引导服务器上运行良好，但嵌入式Kafka从未实现过

@KafkaListener(topics = "${kafka.topic1}", groupId = "groupId")
    public void consume(List<GenericRecord> records, Acknowledgment ack) throws DDCException {
        LOGGER.info("Batch of {} records received", records.size());
        //do something with the data
        ack.acknowledge();
    }

日志中的一切看起来都很好，但我们不知道遗漏了什么。

提前感谢。

Answer 1

我也面临同样的问题与嵌入式Kafka尝试使用KafkaContainer

@ActiveProfiles({"test"})
@RunWith(Cucumber.class)
@CucumberOptions(features= {"src/test/resources/cucumber/data.feature"},
        plugin = {"pretty", "json:target/cucumber.json"})
@SpringBootTest(classes = MyApplication.class)
public final class MyApplicationCucumberTest {
    private MyApplicationCucumberTest() {}

    @Container
    private static KafkaContainer kafkaContainer = new KafkaTestContainer();


    @BeforeClass
    public static void beforeClass() throws IOException, TTransportException {

        kafkaContainer.start();
        System.out.println("Kafka Bootstrap server : " + kafkaContainer.getBootstrapServers());
        System.setProperty("spring.kafka.bootstrap", kafkaContainer.getBootstrapServers());
        System.out.println("Kafka Bootstrap server : " + System.getProperty("spring.kafka.bootstrap"));
        try {
            // Create Topic 
            kafkaContainer.execInContainer("/bin/sh", "-c", "/usr/bin/kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic my.topic");
            
        } catch (InterruptedException e) {
            e.printStackTrace();
            System.out.println(e);
        }

    }


}

还可以添加Spring。Kafka。消费者自动偏移重置：最早的应用。yml

public class KafkaTestContainer extends KafkaContainer {
    private static final String KAFKA_DOCKER = "confluentinc/cp-kafka:5.4.3";

    public KafkaTestContainer() {
        super(DockerImageName.parse(KAFKA_DOCKER));
        

    }
}

Answer 2

你的考试在开始前就结束了；请参见包含0-0-C-1的线程名称；消费者在启动后不到一秒钟就停止了。

我只是检查了一下，没有，我的测试正在执行，因为您可以在日志的第1174行中看到ProducerConfig值的日志。这个日志就在Kafka模板之后出现。发送（主题、实体）。我不使用@Test，因为Cucumber中有step定义。你可以在我的帖子里看到代码。

好啊但是在测试中需要某种锁存，以等待使用者实际分配主题/分区并接收数据。按照您现在构建测试的方式，测试在使用者完全启动之前就已关闭。查看我对这个问题的回答，了解一种包装听众的方法，这样你就可以等到收到录音。（这使用正常的JUnit测试）。

另一种技术是以某种方式将服务注入到侦听器bean中，以倒计时闩锁。

作为快速测试，添加线程。睡眠（10_000）到你的“步骤”。

但是，大概你会想要断言消费者确实得到了数据。你需要在测试退出之前做这个断言，因为它是异步的，你需要一些机制来等待它发生（或者超时）。

Answer 3

问题是消费者没有连接到嵌入式Kafka。您可以通过使用test配置文件运行测试，并将以下内容添加到应用程序测试中来实现这一点。yml。

spring:
  kafka:
    bootstrap-servers: ${spring.embedded.kafka.brokers}

然后，您也不需要定制的consumerProperties、consumerFactory和kafkaListenerContainerFactorybean。Spring boot将为您自动连接这些。如果您确实希望使用这些bean（不知道为什么），那么应该仔细检查KafkaAutoConfiguration，以确保覆盖了正确的名称和类型。