Citrus Cucumber Kafka测试用例仅首次执行成功问题排查

我使用Citrus Framework编写了Cucumber测试用例，用于测试一个内部会向Kafka发送消息的HTTP接口。测试用例采用Given-When-Then语法编写，首次执行时能正常调用HTTP端点并验证Kafka消息接收，但重复执行时总是失败，仅首次成功。

测试用例代码 #

@When("^I make a http call$")
public void when_scenario() {
    runner.when(http()
            .client("myHTTPClient")
            .send()
            .post("/api/endpoint")
            .message()
            .contentType(MediaType.APPLICATION_XML_VALUE)
            .header("X-B3-TraceId", "randomstring")
            .body("some XML payload"));
}

@Then("^a kafka message should be sent$")
public void then_scenario() {
    runner.then(receive()
            .endpoint("myKafkaEndpoint")
            .message()
            .timeout(5000)
            .header("ce_source", "mysourcev1"));
}

客户端端点定义 #

@Bean
public HttpClient myHTTPClient() {
    return CitrusEndpoints
        .http()
            .client()
            .requestUrl("http://localhost:10000")
        .build();
}

@Bean
public KafkaEndpoint myKafkaEndpoint() {
    return CitrusEndpoints
        .kafka()
            .asynchronous()
            .server("localhost:9092")
            .topic("my-topic")
            .offsetReset("earliest")
        .build();
}

报错信息 #

org.citrusframework.exceptions.TestCaseFailedException: Action timeout after 5000 milliseconds. Failed to receive message on endpoint: 'my-topic'

问题核心在于Kafka消费者偏移量的处理逻辑：首次测试后，消费者已经提交了偏移量到topic的最新位置，重复测试时，新消息发送后，消费者会从上次提交的偏移量开始拉取，而非从头开始，导致无法捕获到新产生的消息。以下是几种可行的修复方案：

• 方案1：每次测试使用唯一消费者组ID
  为每个测试实例生成唯一的消费者组ID，这样Kafka会将其视为新消费者，自动应用offsetReset("earliest")配置。修改Kafka端点定义：

  @Bean
  public KafkaEndpoint myKafkaEndpoint() {
      return CitrusEndpoints
          .kafka()
              .asynchronous()
              .server("localhost:9092")
              .topic("my-topic")
              .offsetReset("earliest")
              .consumerGroupId("test-group-" + UUID.randomUUID()) // 生成唯一组ID
          .build();
  }
  

• 方案2：测试前后手动重置偏移量
  在测试的@After方法中，将消费者偏移量重置到topic起始位置，确保下次测试能拉取到最新消息：

  @After
  public void resetKafkaOffset() {
      KafkaEndpoint kafkaEndpoint = applicationContext.getBean(KafkaEndpoint.class);
      kafkaEndpoint.getConsumer().seekToBeginning(Collections.singletonList(new TopicPartition("my-topic", 0)));
  }
  

• 方案3：禁用自动提交偏移量
  配置消费者禁用自动提交，在测试@Before方法中手动将偏移量设置到topic末尾，确保只拉取测试过程中产生的新消息：

  @Bean
  public KafkaEndpoint myKafkaEndpoint() {
      return CitrusEndpoints
          .kafka()
              .asynchronous()
              .server("localhost:9092")
              .topic("my-topic")
              .offsetReset("earliest")
              .autoCommit(false) // 禁用自动提交
          .build();
  }
  
  @Before
  public void setupKafkaConsumer() {
      KafkaEndpoint kafkaEndpoint = applicationContext.getBean(KafkaEndpoint.class);
      kafkaEndpoint.getConsumer().seekToEnd(Collections.singletonList(new TopicPartition("my-topic", 0)));
  }
  

• 额外优化点
  确保HTTP请求每次都能触发Kafka消息发送，避免后端因重复TraceId跳过处理。将静态TraceId改为动态生成：

  .header("X-B3-TraceId", UUID.randomUUID().toString())
  

内容的提问来源于stack exchange，提问作者Vishnukanth