这种场景我之前也碰到过——单独跑Kafka消费者main方法一切正常，一集成到Cucumber步骤里就完全收不到消息，大概率是生命周期、配置或者线程模型的问题，咱们一步步排查解决：

1. 最常见的原因：程序提前退出，消费者没来得及拉取消息 #

Kafka消费者是异步后台拉取消息的，你的main方法里肯定有阻塞逻辑（比如awaitTermination或者无限循环+poll），让主线程等着消费者完成消费。但Cucumber步骤执行完就会继续走后续流程，甚至直接结束测试，消费者线程还没来得及拉取到消息就被JVM终止了。

解决办法：给消费方法加明确的等待逻辑 #

要么设置固定的等待时长，要么等收到指定数量的消息再停止，比如：

public void consumeKafkaMessages() throws InterruptedException {
    Properties props = new Properties();
    props.put("bootstrap.servers", "serv1,serv2,serv3");
    props.put("group.id", "cucumber-test-group");
    props.put("auto.offset.reset", "earliest");
    // 其他必要配置...

    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    consumer.subscribe(Collections.singletonList("your-target-topic"));

    try {
        int receivedCount = 0;
        long startTime = System.currentTimeMillis();
        // 最多等30秒，或者收到5条消息就停止
        while (receivedCount < 5 && System.currentTimeMillis() - startTime < 30000) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(2));
            for (ConsumerRecord<String, String> record : records) {
                // 处理消息的逻辑，比如存到变量供Cucumber断言用
                System.out.printf("收到消息: key=%s, value=%s%n", record.key(), record.value());
                receivedCount++;
            }
        }
    } finally {
        consumer.close();
    }
}

这样保证消费者有足够时间拉取消息，不会被Cucumber的步骤流程直接打断。

2. 消费者组ID或位移配置冲突 #

如果你的main方法用的是固定group.id，而Cucumber调用时重复用了同一个ID，之前的消费位移已经提交到最新位置，消费者启动后自然没有新消息可消费。或者auto.offset.reset默认是latest，但测试时没有新消息产生，也会收不到。

解决办法： #

• 给Cucumber测试用的消费者设置唯一的group.id，比如加上时间戳避免重复：

props.put("group.id", "cucumber-test-" + System.currentTimeMillis());

• 强制拉取历史消息，把auto.offset.reset设为earliest：

props.put("auto.offset.reset", "earliest");

3. Cucumber的生命周期导致消费者被提前销毁 #

Cucumber默认每个步骤都是独立的，有些场景下会在步骤执行后清理上下文，导致消费者实例被提前关闭。另外，Cucumber的执行线程和消费者的后台线程可能存在冲突，导致消费者线程被强制终止。

解决办法：把消费者和Cucumber场景生命周期绑定 #

用Cucumber的@Before和@After注解，在场景开始时初始化消费者，场景结束时再关闭，比如：

public class KafkaConsumerStepDefinitions {
    private KafkaConsumer<String, String> consumer;
    private List<ConsumerRecord<String, String>> receivedMessages = new ArrayList<>();

    @Before
    public void setupKafkaConsumer() {
        Properties props = new Properties();
        // 配置初始化...
        consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("test-topic"));
    }

    @Given("I consume messages from Kafka test topic")
    public void consumeMessages() throws InterruptedException {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10));
        records.forEach(receivedMessages::add);
    }

    @Then("I should see at least {int} messages")
    public void verifyMessageCount(int expectedCount) {
        assert receivedMessages.size() >= expectedCount : "收到的消息数量不足";
    }

    @After
    public void teardownKafkaConsumer() {
        if (consumer != null) {
            consumer.close();
        }
        receivedMessages.clear();
    }
}

这样消费者在整个测试场景的生命周期内都处于活跃状态，不会被提前销毁。

4. 异常被静默吞噬，导致消费者初始化失败 #

main方法里的异常会直接打印到控制台，但Cucumber可能会捕获异常并静默处理，或者你的方法里没有正确抛出异常，导致消费者初始化失败（比如bootstrap地址错误、权限问题）但你完全不知道。

解决办法：添加详细的日志和异常抛出 #

在消费方法里加上异常捕获和日志打印，同时把异常抛出来让Cucumber感知到：

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ConsumerKafka {
    private static final Logger logger = LoggerFactory.getLogger(ConsumerKafka.class);

    public void consume() {
        try {
            // 消费者初始化和消费逻辑
        } catch (Exception e) {
            logger.error("Kafka消费者初始化或消费失败", e);
            // 抛出运行时异常，让Cucumber测试失败并显示错误信息
            throw new RuntimeException("Kafka消费异常", e);
        }
    }
}

这样能快速定位到是网络问题、配置错误还是权限问题。

内容的提问来源于stack exchange，提问作者Don Bosco Joseph