java - Camel 卡夫卡集成

Question

我最近注意到 Camel 现在有自己的 Kafka 组件，所以我决定试一试。

我决定尝试一个不错的简单文件 -> kafka 主题如下...

<route>
        <from uri="file:///tmp/input" />
        <setHeader headerName="kafka.PARTITION_KEY">
            <constant>Test</constant>
        </setHeader>
        <to uri="kafka:localhost:9092?topic=test&amp;zookeeperHost=localhost&amp;zookeeperPort=2181&amp;groupId=group1" />
</route>

这看起来很简单，但是，在运行它时我得到...

    java.lang.ClassCastException: java.lang.String cannot be cast to [B
    at kafka.serializer.DefaultEncoder.toBytes(Encoder.scala:34)
    at org.apache.camel.component.kafka.KafkaProducer.process(KafkaProducer.java:78)

在检查 Camel 代码时，它执行以下操作......

String msg = exchange.getIn().getBody(String.class);
KeyedMessage<String, String> data = new KeyedMessage<String, String>(topic, partitionKey.toString(), msg);
producer.send(data);

显然，这是一个序列化问题，我只是不确定是否有解决方法，或者这是现有实现的固有错误？ (或者希望只是我的误会)

有什么建议吗？谢谢，J

Answer 1

啊，没关系，我们开始...希望这对其他人有帮助，您必须在选项中设置序列化器。

<route>
            <from uri="file:///tmp/input" />
            <setHeader headerName="kafka.PARTITION_KEY">
                <constant>Test</constant>
            </setHeader>
            <to uri="kafka:localhost:9092?topic=test&amp;zookeeperHost=localhost&amp;zookeeperPort=2181&amp;groupId=group1&amp;serializerClass=kafka.serializer.StringEncoder" />
</route>