使用kafkastreams的json，我想将其拆分并发送到两个不同的主题

Question

我有一个来自Kafka主题的大型json，我正在将其转换为Java对象，以提取find DB中所需的值。一些记录中会有一系列缺陷，我需要捕获这些缺陷并将其发送到不同的主题，这样它们就可以在数据库中自己的表中结束。使用接收器连接器将值插入数据库，这就是为什么我们使用多个主题。

我发现了分支和拆分，但这似乎更适合于确定一条已消费的记录应该转到哪个主题，而不是将记录的不同部分发送到不同的主题。有没有办法做到这一点，或者我需要在某个地方改变我的架构。

    @Autowired
    void buildPipeline(StreamsBuilder builder) throws Exception{
        KStream<String, String> messageStream = builder.stream(inputTopic, Consumed.with(STRING_SERDE, STRING_SERDE));
                logger.info("started consumer");
        System.out.println(messageStream.toString());

        KStream<String, String> auditRecords = messageStream
                  .map((key, value) -> {
                      try {
                          return new KeyValue<>("", mapStringToAuditOutput(value));
                      } catch (JsonProcessingException e) {
                          e.printStackTrace();
                      }
                      return null;
                  });
     auditRecords.to(outputTopic);

    }
    public String mapStringToAuditOutput(String input) throws JsonProcessingException {
        ObjectMapper mapper = new ObjectMapper();
        AuditOutput auditResults = null;
        try {
            auditResults=  mapper.readValue(input, AuditOutput.class);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        //check for multiple defects
        if(auditResults.getCaseObject().getDefects() != null){
            //send defects to separate topic

        }
        String auditJson = JsonFlattener.flatten(mapper.writeValueAsString(auditResults));
        System.out.println(auditJson);
        return auditJson;
    }

Answer 1

发现了分支和分裂，但看起来这更多的是为了确定消费记录应该转到哪个主题

对的在分支之前，您需要过滤map/MapValue，以便将部分/整个事件发送到不同的主题

更具体地说，创建中间KStream实例并多次使用to（）