Kafka - Camel 文件读写错误答案

【问题标题】：Kafka - Camel file read and write errorKafka - Camel 文件读写错误
【发布时间】：2016-12-01 04:01:28
【问题描述】：

我正在尝试将 Apache camel 与 Kafka 集成，并编写了一个示例程序来读取文件并写入 Kafka 主题。但是这样做时我遇到了错误。我可以以相反的方式从 Kafka 主题读取并写入文件。

堆栈跟踪

org.apache.kafka.common.errors.SerializationException：无法将类 org.apache.camel.component.file.GenericFile 的值转换为值中指定的类 org.apache.kafka.common.serialization.StringSerializer。串行器 [#0 - file://C:%5Cshare%5Cinput] KafkaProducer WARN 没有设置消息键或分区键 [#0 - file://C:%5Cshare%5Cinput] GenericFileOnCompletion WARN 回滚文件策略：org.apache.camel.component.file.strategy.GenericFileRenameProcessStrategy@7127845b 用于文件：GenericFile[C:\share\input\file.文本] [#0 - file://C:%5Cshare%5Cinput] DefaultErrorHandler 错误 (MessageId: ID-L8-CWBL462-49953-1480494317350-0-21 on ExchangeId: ID-L8-CWBL462-49953-1480494317350-0 -22)。交付尝试后用尽：1 捕获：org.apache.kafka.common.errors.SerializationException：无法将类 org.apache.camel.component.file.GenericFile 的值转换为类 org.apache.kafka.common.serialization。 value.serializer中指定的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);
}

}

【问题讨论】：

标签： apache-camel apache-kafka

【解决方案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);
    }

【讨论】：