如何使用 Spring Cloud Stream 测试 apache kafka 中的状态存储？

Question

我正在尝试为我实现 spring 云流 Kafka 流绑定器的拓扑实现测试代码。我正在使用功能样式，所以我想在商店中测试拓扑结果。但是当我调用时从商店返回 null 。你有什么想法吗？

这是我的测试代码；

@Log4j2
@RunWith(SpringRunner.class)
@SpringBootTest(
    webEnvironment = SpringBootTest.WebEnvironment.NONE,
    properties = {"server.port=0"})
public class ApplicationTests {
    @ClassRule
    public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, 1,
            "input-topic");
    private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka();

    @Autowired
    private QueryService queryService;    

    @Test
    public void SimpleProcessorApplicationTest() {
        //I'm producing data here
            var result = queryService.getFromStore();
            assert (actualResultSet.equals(result));        
    }
}

测试项目的配置文件；

spring:
  cloud:
    stream:
      function:
        definition: event1;event2;event3
      bindings:
        event1-in-0:
          destination: input-topic
          consumer:
            timestampExtractorBeanName: eventTimeExtractor
            dlqName: detail-dlq
        event2-in-0:
          destination: input-topic
          consumer:
            timestampExtractorBeanName: eventTimeExtractor
            dlqName: detail2-dlq
        event3-in-0:
          destination: input-topic
          consumer:
            timestampExtractorBeanName: eventTimeExtractor
            dlqName: detail3-dlq
      kafka:
        streams:
          binder:
            state-store-retry:
              max-attempts: 2
              backoff-period: 1000
            replication-factor: 1
            brokers: ${spring.embedded.kafka.brokers}
            configuration:
              commit.interval.ms: 10000
              state.dir: state-store-test
              application.server: 127.0.0.1:8080
              default:
                key:
                  serde: org.apache.kafka.common.serialization.Serdes$StringSerde
                value:
                  serde: org.apache.kafka.common.serialization.Serdes$StringSerde
            functions:
              event1:
                applicationId: aa-event4
              event2:
                applicationId: aa-event5
              event3:
                applicationId: aa-event6
            deserialization-exception-handler: sendtodlq

是 QueryService 类；

@Log4j2
@Service
public class QueryService {

    @Autowired
    InteractiveQueryService interactiveQueryService;

    public List<KeyValue<Integer, Result>> maxResults(){
        List<KeyValue<Integer, Result>> allResult =new ArrayList<>();
        final List<HostInfo> hostInfoList =
                interactiveQueryService.getAllHostsInfo("m1-store");
        for(HostInfo info: hostInfoList){
            if(info.equals(interactiveQueryService.getCurrentHostInfo())){
                log.info("Retrieving all key/value pairs from Local...");
                allResult.addAll(getAllValues());
            }
            else{
                log.info("Retrieving all key/value pairs from Remote...");
                allResult.addAll(getAllValuesFromRemote(info));
            }
        }
        return allResult;
    }

    private List<KeyValue<Integer, Result>> getAllValues() {
        List<KeyValue<Integer, Result>> results = new ArrayList<>();
        ReadOnlyKeyValueStore<Integer, Result> resultStore = interactiveQueryService.getQueryableStore(
                "m1-store", QueryableStoreTypes.keyValueStore());
        resultStore.all().forEachRemaining(results::add);
        return results;
    }

    private List<KeyValue<Integer, Result>> getAllValuesFromRemote(HostInfo hostInfo){
        String targetHost = String.format("http://%s:%d/dept/local", hostInfo.host(), hostInfo.port());
        RestTemplate restTemplate = new RestTemplate();
        List<KeyValue<Integer, Result>> result = restTemplate.getForObject(targetHost,List.class);
        return result;
    }
}

Answer 1

我发现了几个问题。您在配置中的application.server 属性指向localhost:8080。将其更改为 ${spring.embedded.kafka.brokers}。此外，在您的测试中，您发送到错误的主题名称：template.setDefaultTopic("stage.dashboard.iddaa.coupondetail");。 您在配置中将目的地定义为input-topic。在进行这些更改之后，我能够更进一步，但是，您的 aggregate 呼叫根本没有被调用。我没有机会对此进行调试，但我怀疑这与您的窗口逻辑有关。基本上，您的处理器没有将任何内容放入商店agg-store，这就是对商店的查询返回空的原因。看看为什么会这样。希望这些是您进一步分类的一些指示。