Apache camel kafka 在生成 msg 但丢失标头之前聚合答案

【问题标题】：Apache camel kafka aggregate before produce msg but lost headerApache camel kafka 在生成 msg 但丢失标头之前聚合
【发布时间】：2022-06-17 15:44:25
【问题描述】：

我在 Spring Boot 中使用了 Apache Camel Kafka
3.14.2
我在 apache camel Kafka 组件上使用了默认配置

    <dependency>
        <groupId>org.apache.camel.springboot</groupId>
        <artifactId>camel-kafka-starter</artifactId>
        <version>${camel.version}</version>
    </dependency>

我的路线骆驼 - fileConsume 有 6000 行

from(fileConsume).split(body().tokenize()).setHeader("testHeader", "valueHeader").aggregate(new GroupedMessageAggregationStrategy())
            .constant(true).completionTimeout(100L).to("kafka:topicTest");

Kafka 上生成的文件中的所有消息都非常快（少于 2 秒），但标头不存在。

当我删除聚合时

 from(fileConsume).split(body().tokenize()).setHeader("testHeader", "valueHeader").to("kafka:topicTest");

Kafka 上产生的文件中的所有消息都非常低（超过 10 分钟），但标头存在。

我需要一些帮助来使用 apache camel kafka 组件在带有标头的速度方式上生成消息。

【问题讨论】：

标签： java apache-kafka apache-camel

【解决方案1】：

you must do this, in order to keep header when aggregation is doing.
<pre>
from("sftp://xxxxxxx@localhost:"
+ "2222/data/in"
+ "?password="
+ "&preferredAuthentications=publickey"
+ "&knownHostsFile=~/.ssh/known_hosts"
+ "&privateKeyFile=xxxxxxx"
+ "&privateKeyPassphrase="
+ "&passiveMode=true"
+ "&fastExistsCheck=true"
+ "&download=true"
+ "&delete=true"
+ "&stepwise=false"
+ "&antInclude=*"
+ "&antExclude=**reject**"
+ "&recursive=false"
+ "&maxMessagesPerPoll=10"
+ "&initialDelay=0"
+ "&delay=0"
+ "&connectTimeout=10000"
+ "&soTimeout=300000"
+ "&timeout=30000"
+ "&shuffle=true"
        + "&eagerMaxMessagesPerPoll=false"
        + "&moveFailed=reject"
        + "&binary=true"
        + "&localWorkDirectory=/opt/camel_data/kafka/"
        + "&readLock=none"
        + "&readLockCheckInterval=1000"
        + "&readLockMinLength=1"
        + "&readLockLoggingLevel=INFO"
        + "&readLockIdempotentReleaseDelay=10000"
        + "&readLockRemoveOnCommit=false"
        + "&readLockRemoveOnRollback=true"
        + "&bulkRequests=1000"
        + "&charset=utf-8")
        .routeId("Consume SFTP")
        .id("Consume SFTP")
        .setProperty("yoda_core_technical_id").header(Exchange.BREADCRUMB_ID)
        .setProperty("x_filename_source").header(Exchange.FILE_NAME_ONLY)
        .setProperty("x_filepath_source").header("CamelFileAbsolutePath")
        .setProperty("x_correlation_id").header("CamelFileName")
        .split(body().tokenize())
            .setHeader("test",constant("test"))
        //.end()
        .aggregate(new GroupedMessageAggregationStrategy())
            .constant(true)
            .completionTimeout(100L)
        //.end() this line deactivate cause an error because the aggregator have no outputprocessor
        .to("direct:aggregate");

from("direct:aggregate")
.process(new Processor() {
            @Override
            public void process(Exchange exchange) throws Exception {
                System.out.println(exchange);
                GenericFileMessage<String> message =(GenericFileMessage<String>) exchange.getMessage().getBody(List.class).get(0);
                exchange.getMessage().setHeader("test",
                        message.getHeader("test"));
            }
        })
        .to("mock:result");
</pre>

【讨论】：