4

我们有一个集成测试，我们使用 EmbeddedKafka 并为一个主题生成一条消息，我们的应用程序处理该消息，然后将结果发送到我们使用和断言输出的第二个主题。在 CI 中，这可能有 2/3 的时间有效，但我们会遇到KafkaTestUtils.getSingleRecord抛出异常的情况java.lang.IllegalStateException: No records found for topic（参见下面的 [1]）。

为了尝试解决这个问题，我ContainerTestUtils.waitForAssignment为注册表中的每个侦听器容器添加了（参见下面的 [2]）。在 CI 中成功运行几次后，我看到了一个新异常：java.lang.IllegalStateException: Expected 1 but got 0 partitions. 这让我想知道这是否真的是原始异常未找到记录的根本原因。

有什么想法可以帮助解决这里的随机故障吗？我将不胜感激有关如何进行故障排除的任何建议。

spring-kafka 和 spring-kafka-test v2.6.4。

编辑：添加newConsumer以供参考。

我们的设置示例：

@SpringBootTest
@RunWith(SpringRunner.class)
@DirtiesContext
@EmbeddedKafka(
    topics = { "topic1","topic2" },
    partitions = 1,
    brokerProperties = {"listeners=PLAINTEXT://localhost:9099", "port=9099"})
public class IntegrationTest {

  @Autowired
  private EmbeddedKafkaBroker embeddedKafkaBroker;

  @Autowired
  private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

  @Test
  public void testExample() {
    try (Consumer<String, String> consumer = newConsumer()) {
      for (MessageListenerContainer messageListenerContainer : kafkaListenerEndpointRegistry.getListenerContainers()) {
        [2]
        ContainerTestUtils.waitForAssignment(messageListenerContainer, embeddedKafkaBroker.getPartitionsPerTopic());
      }

      try (Producer<String, String> producer = newProducer()) {
        embeddedKafkaBroker.consumeFromAnEmbeddedTopic(consumer, "topic2"); // [1]

        producer.send(new ProducerRecord<>(
            "topic1",
            "test payload"));
        producer.flush();
      }

      String result = KafkaTestUtils.getSingleRecord(consumer, "topic2").value();
      assertEquals(result, "expected result");
    }
  }

  private Consumer<String, String> newConsumer() {
    Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("groupId", "false", embeddedKafkaBroker);
    ConsumerFactory<String, AssetTransferResponse> consumerFactory = new DefaultKafkaConsumerFactory<>(
        consumerProps,
        new StringDeserializer(),
        new CustomDeserializer<>());
    return consumerFactory.createConsumer();
  }
}
4

0 回答 0