从Kafka到另一个Kafka的阿帕奇骆驼之路

Question

我正在使用Apache Camel使用来自kafka主题的消息，然后处理该消息，同时处理如果发生异常，我将该消息重定向到另一个kafka主题，并以单独的路由处理该消息。所以我有一个如下所示的路线。

from ("kafka1").process("someProcessor").end();
onException(Throwable.class).process(exchange->{exchange.getIn().setBody("Message with error details")}).to("kafka2");

上面的代码实际上是以相同的kafka(kafka1)发送错误消息。

我通过在onexception进程中设置Exchange.getin().setheader(kafkaconstants.topic,“kafka2”）解决了这一问题。这是预期的行为吗？它为什么忽略kafka2而使用kafka1？

> 使用的

版本的camel-2.14.0

消费者：

from("kafka:" + ("kafka.broker") + "?topic="
            + ("offer.kafka.topic")
            + "&zookeeperHost=" + ("kafka.zookeeper.host")
            + "&zookeeperPort=" + ("kafka.zookeeper.port")
            + "&groupId=" + ("offer.kafka.group.id")
            + "&consumerStreams=" + ("kafka.streams")
            + "&autoCommitIntervalMs=" + ("product.kafka.consumer.auto.commit.intervals")
            + "&zookeeperConnectionTimeoutMs=" + ("zookeeper.connection.timeout")
            + "&rebalanceMaxRetries=" + ("kafka.rebalance.max.retries")
            + "&rebalanceBackoffMs=" + ("kafka.rebalance.backoffs.ms")
            + "&zookeeperSessionTimeoutMs=" + ("zookeeper.session.timeout")
            + "&autoOffsetReset=" + ("kafka.auto.offset.reset")
            + "&fetchMessageMaxBytes=" + ("kafka.fetch.message.max.bytes")
            + "&socketReceiveBufferBytes=" + ("receive.buffer.bytes"))
            .routeId("offerEventRoute").to("direct:offerEventRoute");

制作人：

to("kafka:" + ("error.kafka.broker") + "?topic="
                        + ("error.kafka.topic")
                        + "&zookeeperHost=" + ("error.kafka.zookeeper.host")
                        + "&zookeeperPort=" + ("error.kafka.zookeeper.port")
                        + "&groupId=" + ("error.kafka.group.id")
                        + "&zookeeperConnectionTimeoutMs=" + ("error.zookeeper.connection.timeout")
                        + "&rebalanceMaxRetries=" + ("rebalance.max.retries")
                        + "&rebalanceBackoffMs=" + ("rebalance.backoffs.ms")
                        + "&zookeeperSessionTimeoutMs=" + ("zookeeper.session.timeout")
                        + "&autoOffsetReset=" + ("auto.offset.reset")
                        + "&messageSendMaxRetries=" + ("error.max.retries")
                        + "&serializerClass=kafka.serializer.StringEncoder"
        );

Answer 1

您需要在您的producer kafkaendpoint中将bridgeEndPoint设置为true。否则，它将在exchange标头中查找主题名称，并将其用作生产者的主题名称。

默认情况下为false。