kafka-骆驼文件读写错误

Question

我试图将Apache骆驼与Kafka集成在一起，并编写了一个示例程序来读取文件并写入Kafka主题。但是我这样做的时候出错了。我可以用相反的方式从Kafka主题读取并写入文件。

组织。阿帕奇。Kafka。常见的错误。SerializationException:无法转换类org的值。阿帕奇。骆驼组成部分文件将GenericFile发送到类组织。阿帕奇。Kafka。常见的序列化。在值中指定了StringSerializer。序列化程序[#0-file://C:\share\input]KafkaProducer WARN No message key或partition key set[#0-file://C:\share\input]GenericFileOnCompletion WARN回滚文件策略：org。阿帕奇。骆驼组成部分文件策略GenericFileRenameProcessStrategy@7127845b对于文件：GenericFile[C:\share\input\file.txt][#0-file://C:\share\input]DefaultErrorHandler错误传递失败（交换ID:ID-L8-CWBL462-49953-1480494317350-0-21上的MessageId:ID-L8-CWBL462-49953-1480494317350-0-22）。尝试交付后筋疲力尽：1捕获：组织。阿帕奇。Kafka。常见的错误。SerializationException:无法转换类org的值。阿帕奇。骆驼组成部分文件将GenericFile发送到类组织。阿帕奇。Kafka。常见的序列化。在值中指定了StringSerializer。序列化程序

密码

@ContextName（“myCdiCamelContext”）公共类MyRoutes扩展了RouteBuilder{

 @Inject
 @Uri("file:C:\\share\\input?fileName=file.txt&noop=true")
 private Endpoint inputEndpoint;

 @Inject
 @Uri("kafka:localhost:9092?topic=test&groupId=testing&autoOffsetReset=earliest&consumersCount=1")
 private Endpoint resultEndpoint;



@Override
public void configure() throws Exception {
    from(inputEndpoint)
         .to(resultEndpoint);
}

}

Answer 1

添加新处理器后，它对我有效

public void configure() throws Exception {
        from(inputEndpoint).process(new Processor() {
            @Override
            public void process(Exchange exchange) throws Exception {
                exchange.getIn().setBody(exchange.getIn().getBody(),String.class);
                exchange.getIn().setHeader(KafkaConstants.PARTITION_KEY, 0);
                exchange.getIn().setHeader(KafkaConstants.KEY, "1");
            }
        })
             .to(resultEndpoint);
    }