Kafka Streams: определите несколько потоков Kafka с помощью Spring Cloud Stream для каждого набора тем.

Мне удалось решить проблему. Я пишу это для блага других. Если вы хотите включить несколько потоков в единый jar-файл приложения, то ключевым моментом является определение нескольких идентификаторов приложений, по одному на каждый из ваших потоков. Я знал это с самого начала, но не знал, как это определить. Наконец, ответ - это то, что мне удалось найти после прочтения документации SCSt. Ниже показано, как можно определить application.yaml. application.yaml, как показано ниже

spring:
  application.name: kafkaMultiStreamConsumer
  cloud:
    function:
      definition: processAAA; processBBB; processCCC --> // needed for Imperative @StreamListener
    stream:
      kafka: 
        binder:
          brokers: 127.0.0.1:9092
          min-partition-count: 3
          replication-factor: 2
          transaction:
            transaction-id-prefix: transaction-id-2000
          autoCreateTopics: true
          auto-add-partitions: true
        streams:
          binder:
            functions: 
            // needed for functional
              processBBB: 
                application-id: SampleBBBapplication
              processAAA: 
                application-id: SampleAAAapplication
              processCCC: 
                application-id: SampleCCCapplication
            configuration: 
              commit.interval.ms: 1000            
              default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
              default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde        
      bindings:
      // Below is for Imperative Style programming using 
      // the annotation namely @StreamListener, @SendTo in .java class
        inputAAA:
          destination: aaaInputTopic
        outputAAA:
          destination: aaaOutputTopic
        inputBBB:
          destination: bbbInputTopic
        outputBBB:
          destination: bbbOutputTopic
        inputCCC:
          destination: cccInputTopic
        outputCCC:
          destination: cccOutputTopic
     // Functional Style programming using Function<KStream...> use either one of them
     // as both are not required. If you use both its ok but only one of them works
     // from what i have seen @StreamListener is triggered always.
     // Below is from functional style
        processAAA-in-0:
          destination: aaaInputTopic
          group: processAAA-group
        processAAA-out-0:
          destination: aaaOutputTopic
          group: processAAA-group
        processBBB-in-0:
          destination: bbbInputTopic
          group: processBBB-group
        processBBB-out-0:
          destination: bbbOutputTopic
          group: processBBB-group
        processCCC-in-0:
          destination: cccInputTopic
          group: processCCC-group
        processCCC-out-0:
          destination: cccOutputTopic
          group: processCCC-group

Как только выше определено, нам теперь нужно определить отдельные классы Java, в которых реализована логика обработки потока. Ваш класс Java может быть примерно таким, как показано ниже. Создайте аналогично для других 2 или N потоков в соответствии с вашими требованиями. Один из примеров приведен ниже: AAASampleStreamTask.java.

@Component
@EnableBinding(AAASampleChannel.class) // One Channel interface corresponding to in-topic and out-topic
public class AAASampleStreamTask {
    private static final Logger log = LoggerFactory.getLogger(AAASampleStreamTask.class);

    @StreamListener(AAASampleChannel.INPUT)
    @SendTo(AAASampleChannel.OUTPUT)
    public KStream<String, String> processAAA(KStream<String, String> input) {
        input.foreach((key, value) -> log.info("Annotation AAA *Sample* Cloud Stream Kafka Stream processing {}", String.valueOf(System.currentTimeMillis())));
       ...
       // do other business logic
       ...
        return input;
    }
    
    /**
     * Use above or below. Below style is latest startting from ScSt 3.0 if iam not 
     * wrong. 2 different styles of consuming Kafka Streams using SCSt. If we have 
     * both then above gets priority as per my observation
     */     
    /* 
    @Bean
    public Function<KStream<String, String>, KStream<String, String>> processAAA() {
        return input -> input.peek((key, value) -> log.info(
                "Functional AAA *Sample* Cloud Stream Kafka Stream processing : {}", String.valueOf(System.currentTimeMillis())));
       ...
     // do other business logic
       ...
    }
    */
}

Канал необходим, если вы хотите использовать императивное программирование, а не функциональное. AAASampleChannel.java

public interface AAASampleChannel {
    String INPUT = "inputAAA";
    String OUTPUT = "outputAAA";

    @Input(INPUT)
    KStream<String, String> inputAAA();

    @Output(OUTPUT)
    KStream<String, String> outputAAA();
}

Похоже, вы смешиваете Spring Cloud Stream и Spring Kafka в приложении. При использовании связующего вам не нужно напрямую определять компоненты, требуемые Spring Kafka, такие как KafkaStreams и Topology, скорее они создаются SCSt неявно. Можете ли вы удалить следующие бобы и попробовать еще раз?

@Bean
public KafkaStreams kafkaStreams(KafkaProperties kafkaProperties) {

а также

@Bean
public Topology kafkaStreamTopology() {

Если вы по-прежнему сталкиваетесь с проблемами, поделитесь небольшой выборкой, которую можно воспроизвести, чтобы мы могли провести дальнейшую сортировку.