Связка потоков Spring Cloud Пользовательские заголовки Kafka не принимаются потребителем - PullRequest
Новая
       67

Связка потоков Spring Cloud Пользовательские заголовки Kafka не принимаются потребителем

0 голосов
/ 24 мая 2019

Я использую связыватель Spring Cloud Stream Kafka, выпуск Edgware.SR4.

Я установил настраиваемые заголовки для полезной нагрузки сообщения и опубликовал его, но я не вижу эти заголовки на стороне потребителя.

Я использовал объект сообщения для привязки полезной нагрузки и заголовков.Я попытался добавить свойство spring.cloud.stream.kafka.binder.headers, но оно не сработало

Производитель:

Application.yml 

spring:
  cloud:
    stream:
      bindings:
        sampleEvent:
          destination: sample-event
          content-type: application/json
      kafka:
        binder:
          brokers: localhost:9092
          zkNodes: localhost:2181
          autoCreateTopics: false
          zkConnectionTimeout: 36000

MessageChannelConstants.java 

public class MessageChannelConstants {
    public static final String SAMPLE_EVENT = "sampleEvent";
        private MessageChannelConstants() {}
}

SampleMessageChannels.java 

public interface SampleMessageChannels {

    @Output(MessageChannelConstants.SAMPLE_EVENT)
    MessageChannel sampleEvent();
}

SampleEventPublisher.java 

@Service
@EnableBinding(SampleMessageChannels.class)
public class SampleEventPublisher{

@Autowired
private SampleMessageChannels sampleMessageChannels;

public void publishSampleEvent(SampleEvent sampleEvent) {
        final Message<SampleEvent> message = MessageBuilder.withPayload(sampleEvent).setHeader("appId", "Demo").build();
    MessageChannel messageChannel = SampleMessageChannels.sampleEvent();
    if (messageChannel != null) {
        messageChannel.send(message);
    }
}
}

Потребитель:

application.yml 

spring:
  cloud:
    stream:
      bindings:
        sampleEvent:
          destination: sample-event
          content-type: application/json
      kafka:
        binder:
          brokers: localhost:9092
          zkNodes: localhost:2181
          autoCreateTopics: false
          zkConnectionTimeout: 36000

MessageChannelConstants.java 

public class MessageChannelConstants {

    public static final String SAMPLE_EVENT = "sampleEvent";

    private MessageChannelConstants() {}
}

SampleMessageChannels.java 

public interface SampleMessageChannels {

    @Output(MessageChannelConstants.SAMPLE_EVENT)
    MessageChannel sampleEvent();
}

SampleEventListener.java 

@Service
@EnableBinding(SampleMessageChannels.class)
public class SampleEventListener{

@StreamListener(MessageChannelConstants.SAMPLE_EVENT)
public void listenSampleEvent(@Payload SampleEvent event,
    @Header(required = true, name = "appId") String appId) {

     // do something 
}

Ниже приведено исключение, которое я получил, 

org.springframework.messaging.MessageHandlingException: Missing header 'appId' for method parameter type [class java.lang.String]
    at org.springframework.messaging.handler.annotation.support.HeaderMethodArgumentResolver.handleMissingValue(HeaderMethodArgumentResolver.java:100)
    at org.springframework.messaging.handler.annotation.support.AbstractNamedValueMethodArgumentResolver.resolveArgument(AbstractNamedValueMethodArgumentResolver.java:103)
    at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:112)
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:135)
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:107)
    at org.springframework.cloud.stream.binding.StreamListenerMessageHandler.handleRequestMessage(StreamListenerMessageHandler.java:55)
    at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:109)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127)
    at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:148)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:121)
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:89)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:425)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:375)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105)
    at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:360)
    at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:271)
    at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:188)
    at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:115)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127)
    at org.springframework.integration.channel.FixedSubscriberChannel.send(FixedSubscriberChannel.java:70)
    at org.springframework.integration.channel.FixedSubscriberChannel.send(FixedSubscriberChannel.java:64)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105)
    at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:188)
    at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter.access$200(KafkaMessageDrivenChannelAdapter.java:63)
    at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener.onMessage(KafkaMessageDrivenChannelAdapter.java:372)
    at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener.onMessage(KafkaMessageDrivenChannelAdapter.java:352)
    at org.springframework.kafka.listener.adapter.RetryingAcknowledgingMessageListenerAdapter$1.doWithRetry(RetryingAcknowledgingMessageListenerAdapter.java:79)
    at org.springframework.kafka.listener.adapter.RetryingAcknowledgingMessageListenerAdapter$1.doWithRetry(RetryingAcknowledgingMessageListenerAdapter.java:73)
    at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:287)
    at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:180)
    at org.springframework.kafka.listener.adapter.RetryingAcknowledgingMessageListenerAdapter.onMessage(RetryingAcknowledgingMessageListenerAdapter.java:73)
    at org.springframework.kafka.listener.adapter.RetryingAcknowledgingMessageListenerAdapter.onMessage(RetryingAcknowledgingMessageListenerAdapter.java:39)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:792)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:736)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.access$2100(KafkaMessageListenerContainer.java:246)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer$ListenerInvoker.run(KafkaMessageListenerContainer.java:1025)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.lang.Thread.run(Thread.java:748)

Примечание: я также использую зависимость от весеннего облака и zipkin.

1 Ответ

0 голосов
/ 28 мая 2019

В Edgware (SCSt Ditmars) вы должны указать, какие заголовки будут передаваться.

См. Свойства Kinder для Binder .

Это потому, что Edgware был основан наДо этого Кафка изначально поддерживал заголовки, и мы кодировали заголовки в полезную нагрузку.

spring.cloud.stream.kafka.binder.headers

Списокпользовательские заголовки, которые будут транспортироваться связывателем.

По умолчанию: пусто.

Вы также должны обязательно обновить spring-kafka до версии 1.3.9.RELEASE и клиенты kafka до0.11.0.2.

Желательно, однако, перейти на Finchley или Greemwich.Эти версии изначально поддерживают заголовки.

...