从kafka到jms的camel路由不起作用(jms->kafka->jms)

q3aa0525  于 2021-06-04  发布在  Kafka
关注(0)|答案(1)|浏览(654)

在camel router.java中,我有下一个路由

from("jms:topic:test.source.topic?asyncConsumer=true")
            .log("Message: ${body}")
            .to("kafka:testing?brokers=192.168.0.100:9092");

from("kafka:testing?brokers=192.168.0.100:9092")
            .log("Message received from Kafka : ${body}")
            .log("    on the topic ${headers[kafka.TOPIC]}")
            .log("    on the partition ${headers[kafka.PARTITION]}")
            .log("    with the offset ${headers[kafka.OFFSET]}")
            .log("    with the key ${headers[kafka.KEY]}")
            // manually set JMSDeliveryMode (1 - NON_PERSISTENT, 2 - PERSISTENT)
            .process(new Processor() {
                @Override
                public void process(Exchange exchange) throws Exception {
                    exchange.getIn().setHeader("JMSDeliveryMode", "1");
               }
            })
            .to("jms:topic:test.sink.topic");

上述驼峰路线出现问题。如果我发一些信息到主题 testing 和Kafka的制作人在一起 bin/kafka-console-producer.sh --topic testing --bootstrap-server localhost:9092 ,从kafka到jms的路由工作正常。因此,这是一些与 Camel 路线相连的问题。
在camel pom.xml中有spring boot camel kafka starter和camel jms starter依赖项。
当我用maven启动spring boot camel并从kafka生产者向kafka代理发送一些消息时 testing 主题,我可以看到消息是由kafka代理接收的,上面的日志打印正常。
第行出现错误 .to("jms:topic:test.sink.topic"); 我不知道这是什么意思。

ERROR 28642 --- [aConsumer[testing]] o.a.c.p.e.DefaultErrorHandler: Failed delivery for (MessageId: ID-PCID on ExchangeId: ID-PCID). Exhausted after delivery attempt: 1 caught: org.springframework.jms.UncategorizedJmsException: Uncategorized exception occurred during JMS processing; nested exception is javax.jms.JMSException: AMQ139015: Illegal deliveryMode value: 0

org.springframework.jms.UncategorizedJmsException: Uncategorized exception occurred during JMS processing; nested exception is javax.jms.JMSException: AMQ139015: Illegal deliveryMode value: 0
    at org.springframework.jms.support.JmsUtils.convertJmsAccessException(JmsUtils.java:311) ~[spring-jms-5.2.6.RELEASE.jar:5.2.6.RELEASE]
    at org.springframework.jms.support.JmsAccessor.convertJmsAccessException(JmsAccessor.java:185) ~[spring-jms-5.2.6.RELEASE.jar:5.2.6.RELEASE]
    at org.springframework.jms.core.JmsTemplate.execute(JmsTemplate.java:507) ~[spring-jms-5.2.6.RELEASE.jar:5.2.6.RELEASE]
    at org.apache.camel.component.jms.JmsConfiguration$CamelJmsTemplate.send(JmsConfiguration.java:525) ~[camel-jms-3.4.0.jar:3.4.0]
    at org.apache.camel.component.jms.JmsProducer.doSend(JmsProducer.java:438) ~[camel-jms-3.4.0.jar:3.4.0]
    at org.apache.camel.component.jms.JmsProducer.processInOnly(JmsProducer.java:392) ~[camel-jms-3.4.0.jar:3.4.0]
    at org.apache.camel.component.jms.JmsProducer.process(JmsProducer.java:155) ~[camel-jms-3.4.0.jar:3.4.0]
    at org.apache.camel.processor.SendProcessor.process(SendProcessor.java:168) ~[camel-base-3.4.0.jar:3.4.0]
    at org.apache.camel.processor.errorhandler.RedeliveryErrorHandler$SimpleTask.run(RedeliveryErrorHandler.java:395) ~[camel-base-3.4.0.jar:3.4.0]
    at org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:148) ~[camel-base-3.4.0.jar:3.4.0]
    at org.apache.camel.impl.engine.DefaultReactiveExecutor.scheduleMain(DefaultReactiveExecutor.java:60) ~[camel-base-3.4.0.jar:3.4.0]
    at org.apache.camel.processor.Pipeline.process(Pipeline.java:147) ~[camel-base-3.4.0.jar:3.4.0]
    at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:286) ~[camel-base-3.4.0.jar:3.4.0]
    at org.apache.camel.impl.engine.DefaultAsyncProcessorAwaitManager.process(DefaultAsyncProcessorAwaitManager.java:83) ~[camel-base-3.4.0.jar:3.4.0]
    at org.apache.camel.support.AsyncProcessorSupport.process(AsyncProcessorSupport.java:40) ~[camel-support-3.4.0.jar:3.4.0]
    at org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:346) ~[camel-kafka-3.4.0.jar:3.4.0]
    at org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:222) ~[camel-kafka-3.4.0.jar:3.4.0]
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[na:na]
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[na:na]
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[na:na]
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[na:na]
    at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]
Caused by: javax.jms.JMSException: AMQ139015: Illegal deliveryMode value: 0
    at org.apache.activemq.artemis.jms.client.ActiveMQMessage.setJMSDeliveryMode(ActiveMQMessage.java:450) ~[artemis-jms-client-2.12.0.jar:2.12.0]
    at org.apache.camel.component.jms.JmsMessageHelper.setJMSDeliveryMode(JmsMessageHelper.java:435) ~[camel-jms-3.4.0.jar:3.4.0]
    at org.apache.camel.component.jms.JmsBinding.appendJmsProperty(JmsBinding.java:393) ~[camel-jms-3.4.0.jar:3.4.0]
    at org.apache.camel.component.jms.JmsBinding.appendJmsProperties(JmsBinding.java:371) ~[camel-jms-3.4.0.jar:3.4.0]
    at org.apache.camel.component.jms.JmsBinding.makeJmsMessage(JmsBinding.java:346) ~[camel-jms-3.4.0.jar:3.4.0]
    at org.apache.camel.component.jms.JmsProducer$2.createMessage(JmsProducer.java:325) ~[camel-jms-3.4.0.jar:3.4.0]
    at org.apache.camel.component.jms.JmsConfiguration$CamelJmsTemplate.doSendToDestination(JmsConfiguration.java:561) ~[camel-jms-3.4.0.jar:3.4.0]
    at org.apache.camel.component.jms.JmsConfiguration$CamelJmsTemplate.lambda$send$0(JmsConfiguration.java:527) ~[camel-jms-3.4.0.jar:3.4.0]
    at org.springframework.jms.core.JmsTemplate.execute(JmsTemplate.java:504) ~[spring-jms-5.2.6.RELEASE.jar:5.2.6.RELEASE]
    ... 19 common frames omitted

当从jms向kafka发送一些消息时,路由工作正常。

zphenhs4

zphenhs41#

以下是相关的错误消息:

AMQ139015: Illegal deliveryMode value: 0

此错误消息表示有东西调用了jms方法 javax.jms.Message#setJMSDeliveryMode 值为 0 . 价值 0 无效。有效的传递模式值由定义 javax.jms.DeliveryMode : 1 (对于非持久性) 2 (对于持久性)。
堆栈跟踪表示此无效值是由camel根据输入消息设置的。请参阅此代码。为了解决这个问题,您需要确定标题的位置 JMSDeliveryMode 正在设置为 org.apache.camel.Message .

相关问题