Блог

Я хочу прослушать две темы Kafka, как в приведенном ниже коде, и есть два исходных события, которые необходимо обработать и преобразовать в другое событие.

Итак, я хочу прослушать эти два события в одном EnableBinding

     @EnableBinding(PartnerOrderCancelledEventImporter.Targets.class)
    public class PartnerOrderCancelledEventImporter {

       @StreamListener(Targets.INPUT_ORDER_CANCELLED)
        @SendTo(Targets.OUTPUT)
        public KStream<?, TriggeringEvent> processOrderCancelled(KStream<?, OrderCancelledV1> input) {
            LogInfo("PartnerOrderCancelled-OrderCancelledV1 stream started");
            LogInfo("  KafkaBrokers: "   KafkaBrokers);

            return input
                    .filter((key, value) -> IsFFFaultAndNoRoutes(value))
                    .peek((key, value) -> LogInfo("OrderCancelle", new LogObject(value)))
                    .map((key, value) -> KeyValue.pair(key, new TriggeringEvent(value)));
        }

        @StreamListener(Targets.INPUT_ORDER_ITEM_STOCK_CHECKED)
        @SendTo(Targets.OUTPUT)
        public KStream<?, TriggeringEvent> processOrderItemStockChecked(KStream<?, OrderItemStockCheckedV1> input) {
            LogInfo("PartnerOrderCancelled-OrderItemStockCheckedV1 stream started");
            LogInfo("  KafkaBrokers: "   KafkaBrokers);

            return input
                    .filter((key, value) -> IsItemNoStock(value))
                    .peek((key, value) -> LogInfo("OrderItemStockChecke", new LogObject(value)))
                    .map((key, value) -> KeyValue.pair(key, new TriggeringEvent(value)));
        }


        public interface Targets {

            String INPUT_ORDER_CANCELLED = "partnerOrderCancelledInputOrderCancelled";
            String INPUT_ORDER_ITEM_STOCK_CHECKED = "partnerOrderCancelledInputOrderItemStockChecked";
            String OUTPUT = "triggeringEventsOutputPartnerOrderCancelled";

            @Input(INPUT_ORDER_CANCELLED)
            KStream<?, ?> inputOrderCancelled();

            @Input(INPUT_ORDER_ITEM_STOCK_CHECKED)
            KStream<?, ?> inputOrderItemStockChecked();

            @Output(OUTPUT)
            KStream<?, ?> output();
        }
   spring.cloud.stream.bindings.triggeringEventsOutputPartnerOrderCancelled.destination=dev.comms.triggeringevents.TriggeringEvent-events-1.0
    spring.cloud.stream.bindings.partnerOrderCancelledInputOrderItemStockChecked.destination=dev.ecom.order-management-service.order-item-stock-checked-events-v1
    spring.cloud.stream.bindings.partnerOrderCancelledInputOrderItemStockChecked.group=TriggeringEvents-PartnerOrderCancelled
    spring.cloud.stream.bindings.partnerOrderCancelledInputOrderCancelled.destination=dev.ecom.order-management-service.order-cancelled-events-v1
    spring.cloud.stream.bindings.partnerOrderCancelledInputOrderCancelled.group=TriggeringEvents-PartnerOrderCancelled
  

приведенные ниже исключения

 org.springframework.beans.factory.BeanInitializationException: Cannot setup StreamListener for public org.apache.kafka.streams.kstream.KStream com.farfetch.communication.triggeringeventsimporter.PartnerOrderCancelledEventImporter.processOrderItemStockChecked(org.apache.kafka.streams.kstream.KStream); nested exception is java.lang.reflect.UndeclaredThrowableException

    at org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsStreamListenerSetupMethodOrchestrator.orchestrateStreamListenerSetupMethod(KafkaStreamsStreamListenerSetupMethodOrchestrator.java:195)
    at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.doPostProcess(StreamListenerAnnotationBeanPostProcessor.java:195)
    at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.lambda$postProcessAfterInitialization$0(StreamListenerAnnotationBeanPostProcessor.java:167)
    at java.lang.Iterable.forEach(Iterable.java:75)
    at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.injectAndPostProcessDependencies(StreamListenerAnnotationBeanPostProcessor.java:285)
    at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.afterSingletonsInstantiated(StreamListenerAnnotationBeanPostProcessor.java:105)
    。。。。。。。。。。。。。。。。。。。。。。。。。。。。。
    at org.springframework.cloud.stream.binder.kafka.streams.KStreamStreamListenerResultAdapter.adapt(KStreamStreamListenerResultAdapter.java:41)
    at org.springframework.cloud.stream.binder.kafka.streams.KStreamStreamListenerResultAdapter.adapt(KStreamStreamListenerResultAdapter.java:31)
    at org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsStreamListenerSetupMethodOrchestrator.orchestrateStreamListenerSetupMethod(KafkaStreamsStreamListenerSetupMethodOrchestrator.java:187)
    ... 39 more
Caused by: java.lang.reflect.InvocationTargetException
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.springframework.cloud.stream.binder.kafka.streams.KStreamBoundElementFactory$KStreamWrapperHandler.invoke(KStreamBoundElementFactory.java:99)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185)
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:212)
    ... 43 more
Caused by: java.lang.IllegalArgumentException: delegate already set to org.apache.kafka.streams.kstream.internals.KStreamImpl@491cafec
  

Вы получаете это исключение, потому что пытаетесь связать один и тот же исходящий KStream с обоих процессоров (вывод). Есть ли шанс, что вы можете добавить другую привязку вывода для второго процессора? Это должно устранить эту конкретную ошибку, которую вы получаете. С другой стороны, если это требование, которого требует ваш вариант использования, то в настоящее время binder это не поддерживает (хотя это могло бы быть хорошей возможностью для добавления). В качестве обходного пути вы могли бы заставить второй процессор ничего не возвращать, а затем вызвать to метод на KStream , чтобы отправить его вручную.