Spring Kafka-Kafka流拓扑的建立

Question

我们有一个kafka streams Spring Boot应用程序（使用spring-kafka)，这个应用程序目前从上游主题读取消息，应用一些转换，并将它们写入下游主题，它不做任何聚合或联接或任何高级kafka streams功能。

代码当前类似于

@Bean 
fun topology(streamsBuilder: StreamsBuilder): KStream<String, SomeObject> {
  val stream = streamsBuilder.stream<String, SomeObject>(inputTopicName)
  val branches: Array<KStream<String, SomeObject>> = stream.branch(
    { _, value -> isValidRawData(value)},
    { _, failedValue -> true}
  )
        
  branches[0].map { _, value -> transform(value) }.to(outputTopicName)
  branches[1].foreach { _, value -> s3Service.uploadEvent(value) }
}

@Bean 
fun topology(streamsBuilder: StreamsBuilder): KStream<String, SpecificRecord> {
  val topics = listOf("topic1", "topic2")
  val stream = streamsBuilder.stream<String, SpecificRecord>(topics)
  val branches: Array<KStream<String,SpecificRecord>> = stream.branch(
    { _, value -> isRecordFromTopic1(value)},
    { _, value -> isRecordFromTopic2(value)},
    { _, failedValue -> true}
  )
        
  branches[0].map { _, value -> transformTopic1Record(value) }.to(outputTopicName)
  branches[1].map { _, value -> transformTopic2Record(value) }.to(outputTopicName)
  branches[2].foreach { _, value -> s3Service.uploadEvent(value) }
}

Answer 1

由于如第二段代码所示，有一组主题API，所以我想说这两个变体都是有效的，也是有意义的。其他一切都只是个人喜好。我会选择第一个，因为从技术上来说，最终所有的东西都将在同一个流引擎上工作。第一种解决方案在将来引入第三种记录类型时更容易支持。或者您可能有特定流的额外逻辑。您可能有一个公共流来读取所有主题，并通过该条件和分支分发它们。其余的逻辑可以通过他们自己的中间主题在他们各自的流中完成。但仍然：只是我的观点...