0

ソースとしてkafkaトピックを試すと、kafka例外が発生します

これが私がストリームを作成した方法です

stream create --definition ":myDestination > log" --name ingest_from_broker
stream deploy ingest_from_broker --properties "spring.cloud.stream.bindings.input.consumer.headerMode=raw"

実行中にログファイルで例外が発生します

java.lang.StringIndexOutOfBoundsException: String index out of range: 113
    at java.lang.String.checkBounds(String.java:385) ~[na:1.8.0_66]
    at java.lang.String.<init>(String.java:425) ~[na:1.8.0_66]
    at org.springframework.cloud.stream.binder.EmbeddedHeadersMessageConverter.oldExtractHeaders(EmbeddedHeadersMessageConverter.java:132) ~[spring-cloud-stream-1.0.2.RELEASE.jar!/:1.0.2.RELEASE]
    at org.springframework.cloud.stream.binder.EmbeddedHeadersMessageConverter.extractHeaders(EmbeddedHeadersMessageConverter.java:105) ~[spring-cloud-stream-1.0.2.RELEASE.jar!/:1.0.2.RELEASE]
    at org.springframework.cloud.stream.binder.AbstractBinder.extractMessageValues(AbstractBinder.java:153) ~[spring-cloud-stream-1.0.2.RELEASE.jar!/:1.0.2.RELEASE]
    at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder$ReceivingHandler.handleRequestMessage(KafkaMessageChannelBinder.java:698) [spring-cloud-stream-binder-kafka-1.0.2.RELEASE.jar!/:1.0.2.RELEASE]
    at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:99) [spring-integration-core-4.2.4.RELEASE.jar!/:na]
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127) [spring-integration-core-4.2.4.RELEASE.jar!/:na]
    at org.springframework.integration.channel.FixedSubscriberChannel.send(FixedSubscriberChannel.java:69) [spring-integration-core-4.2.4.RELEASE.jar!/:na]
    at org.springframework.integration.channel.FixedSubscriberChannel.send(FixedSubscriberChannel.java:63) [spring-integration-core-4.2.4.RELEASE.jar!/:na]
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115) [spring-messaging-4.2.7.RELEASE.jar!/:4.2.7.RELEASE]
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45) [spring-messaging-4.2.7.RELEASE.jar!/:4.2.7.RELEASE]
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105) [spring-messaging-4.2.7.RELEASE.jar!/:4.2.7.RELEASE]
    at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:105) [spring-integration-core-4.2.4.RELEASE.jar!/:na]
    at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter.access$300(KafkaMessageDrivenChannelAdapter.java:43) [spring-integration-kafka-1.3.1.RELEASE.jar!/:na]
    at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$AutoAcknowledgingChannelForwardingMessageListener.doOnMessage(KafkaMessageDrivenChannelAdapter.java:171) [spring-integration-kafka-1.3.1.RELEASE.jar!/:na]
    at org.springframework.integration.kafka.listener.AbstractDecodingMessageListener.onMessage(AbstractDecodingMessageListener.java:50) [spring-integration-kafka-1.3.1.RELEASE.jar!/:na]
    at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder$4$1.doWithRetry(KafkaMessageChannelBinder.java:516) [spring-cloud-stream-binder-kafka-1.0.2.RELEASE.jar!/:1.0.2.RELEASE]
    at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:276) [spring-retry-1.1.3.RELEASE.jar!/:na]
    at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:157) [spring-retry-1.1.3.RELEASE.jar!/:na]
    at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder$4.onMessage(KafkaMessageChannelBinder.java:513) [spring-cloud-stream-binder-kafka-1.0.2.RELEASE.jar!/:1.0.2.RELEASE]
    at org.springframework.integration.kafka.listener.QueueingMessageListenerInvoker$KafkaMessageDispatchingSubscriber.onNext(QueueingMessageListenerInvoker.java:221) [spring-integration-kafka-1.3.1.RELEASE.jar!/:na]
    at org.springframework.integration.kafka.listener.QueueingMessageListenerInvoker$KafkaMessageDispatchingSubscriber.onNext(QueueingMessageListenerInvoker.java:209) [spring-integration-kafka-1.3.1.RELEASE.jar!/:na]
    at reactor.core.processor.util.RingBufferSubscriberUtils.route(RingBufferSubscriberUtils.java:67) [reactor-core-2.0.8.RELEASE.jar!/:na]
    at reactor.core.processor.RingBufferProcessor$BatchSignalProcessor.run(RingBufferProcessor.java:789) [reactor-core-2.0.8.RELEASE.jar!/:na]
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_66]
    at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_66]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [na:1.8.0_66]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [na:1.8.0_66]
    at java.lang.Thread.run(Thread.java:745) [na:1.8.0_66]

このプロパティを設定すると言及されているいくつかのスレッドを見ました-「spring.cloud.stream.bindings.input.consumer.headerMode = raw」が役立ちますが、どういうわけかこれは機能しません。

4

1 に答える 1

0

どちらかを試してください

stream create --definition ":myDestination > log --spring.cloud.stream.bindings.input.consumer.headerMode=raw" --name ingest_from_broker

また

stream deploy ingest_from_broker --properties "apps.log.spring.cloud.stream.bindings.input.consumer.headerMode=raw"

つまり、Stream の定義中にプロパティをアプリケーションで直接指定するか、展開時に指定する場合は、どのアプリケーションに適用するかを示す必要があります。

于 2016-08-02T07:53:49.473 に答える