Функциональная модель Spring-cloud-stream с apache-kafka-binder

Question

Это продолжение этого вопроса . Могу ли я использовать «простой» Apache Kafka Binder вместе с функциональной моделью? До сих пор, используя конфигурацию на основе аннотаций, я смешал оба: spring-cloud-stream-binder-kafka для простого потребления / производства и spring-cloud-stream-binder-kafka-streams для расширенной потоковой обработки в одном приложении.

Функциональная модель, кажется, поддерживается только связывателем streams, иесли я пытаюсь смешать оба подхода - аннотацию, основанную на простом использовании и функциональную для потоков, привязка к потоку не регистрируется.

spring.cloud:
        stream:
          function:
            definition: processStream
          bindings:
            processStream-in-0:
              destination:  my-topic
            simple-binding-in:
              destination: another-topic

public interface SimpleBinding {

    String INPUT = "simple-binding-in";

    @Input(INPUT)
    SubscribableChannel simpleIn();

}

@Component
public class SimpleListener {

    @StreamListener(SimpleBinding.INPUT)
    public void listen(@Payload SomeDto payload) {
    }
}

@Configuration
public class FunctionalStream {

    @Bean
    public Consumer<KStream<String>> processStream() {
        return eventStream -> eventStream.map()
    }
}

@EnableBinding(SimpleBinding.class) присутствует в классе конфигурации. Является ли предпочтительным / поддерживается смешивать оба, как описано, или я должен использовать streams-binder даже для простого потребления сообщений?

Answer 1

Для Kafka Binder вы можете и обязательно должны использовать функциональную модель и забыть о StreamListener вместе. Таким образом, он будет приведен в соответствие с вашей функциональной моделью KStream.

spring.cloud:
        stream:
          function:
            definition: processStream
          bindings:
            processStream-in-0:
              destination:  my-topic
            listen-in-0:
              destination: another-topic

@Component
public class SimpleListener {

    @Bean
    public Consumer<SomeDto> listen() {
        return payload -> ...
    }
}

@Configuration
public class FunctionalStream {

    @Bean
    public Consumer<KStream<String>> processStream() {
        return eventStream -> eventStream.map()
    }
}