骆驼kafka主题中的spring-cloud-contract-verifier清理消息答案

【问题标题】：spring-cloud-contract-verifier cleanup messages in camel kafka topics骆驼kafka主题中的spring-cloud-contract-verifier清理消息
【发布时间】：2021-07-28 18:13:01
【问题描述】：

在合约测试期间，我运行 main FLOW，它在不同的 kafka 主题（TOPIC_1 和 TOPIC_2）中生成 2 个事件。我有两个不同的测试来检查这个事件的发送（TEST_1 用于TOPIC_1 和TEST_2 用于TOPIC_2）。所以TEST_1 和TEST_2 运行相同的FLOW，对于TEST_1，我有将事件发送到TOPIC_2 的副作用，对于TEST_2 - TOPIC_1。考虑示例，我运行TEST_1，然后运行TEST_2。在TEST_2 期间，我将在TOPIC_2 内进行2 个事件- 一个由TEST_1 制作，第二个由TEST_2 制作。当然，我的TEST_2 会失败，因为在验证过程中它应该接收到由他自己生成的消息，没有别的。

所以，这就是为什么我需要在每次测试之前跳过所有主题中的所有旧消息。怎么可能使用org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierMessaging

【问题讨论】：

标签： java apache-kafka apache-camel spring-cloud-contract

【解决方案1】：

我找到了一个解决方案，可以轮询来自 CamelContext 内每个端点的所有消息

@Autowired
private org.apache.camel.CamelContext camelContext;

@org.junit.Before
public void cleanUpCamelEndpoints() {
    for (Endpoint endpoint : camelContext.getEndpoints()) {
        try {
            PollingConsumer pollingConsumer = endpoint.createPollingConsumer();
            Exchange exchangeToSkip;
            while ((exchangeToSkip = pollingConsumer.receiveNoWait()) != null) {
                log.debug("Skipped side effect exchange: {}", exchangeToSkip);
            }
        } catch (Exception exception) {
            log.debug("Exception while receive exchange to skip: " + exception);
        }
    }
}

【讨论】：