我想聚合3个主题,但为了聚合策略的poc,我尝试聚合2个主题。
因此,我配置了以下驼峰路由:
@Override
public void configure() throws Exception {
CamelContext contexto = new DefaultCamelContext();
try{
onException(SQLException.class, Exception.class)
.handled(true)
.maximumRedeliveries(1)
.process(new CustomExceptionProcess());
errorHandler(deadLetterChannel("file:data/error") // There is no file created
.maximumRedeliveries(3)
.redeliveryDelay(5000));
from("kafka:topic1")
.process(new ProcessTopic1()) // inside this process
.to("direct:agregador");
from("kafka:topic2")
.process(new ProcessTopic2())
.to("direct:agregador");
from("direct:agregador")
.aggregate(header("idAgregador"), new EstrategiaAgregador()) // implements AggregationStrategy
.completionPredicate(predicadoTamanhoMinimo()) // This just to check if is there at least 2 objects before realease the aggregation.
.eagerCheckCompletion()
.to("direct:insereBanco")
;
from("direct:insereBanco")
.process(new ProcessoInsere());
}catch (Exception e){
LOG.error("Error in the route: ", e);
}
主题1和主题2的过程是这样的:
// It's the same process for ProcessTopic1 and ProcessTopic2
@Override
public void process(Exchange exchange) throws Exception {
if(!(exchange.getIn().getBody() instanceof Object1)){
String someData = (String) exchange.getIn().getBody();
JSONParser parser = new JSONParser();
Object objeto = parser.parse(someData);
JSONObject json = (JSONObject) objeto;
ObjectMapper mapeadorObj = new ObjectMapper();
Object1 someEntity = mapeadorObj.readValue(json.toString(), Object1.class);
ProducerTemplate producerTemplate = exchange.getContext().createProducerTemplate();
producerTemplate.sendBodyAndHeader("direct:agregador", someEntity, "idAgregador", someEntity.getId());
}
}
但是,当我得到要聚合的第一条消息时,消息离开了(因为必须等待第二条消息),但是很快这个消息退出了,我得到了这个异常:
org.apache.camel.CamelExchangeException: Invalid correlation key. Exchange[ID-XXXX-1597437627634-0-1]
at org.apache.camel.processor.aggregate.AggregateProcessor.doProcess(AggregateProcessor.java:302)
at org.apache.camel.processor.aggregate.AggregateProcessor.process(AggregateProcessor.java:271)
at org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:76)
at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:548)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:201)
at org.apache.camel.processor.Pipeline.process(Pipeline.java:138)
at org.apache.camel.processor.Pipeline.process(Pipeline.java:101)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:201)
at org.apache.camel.component.direct.DirectProducer.process(DirectProducer.java:76)
>> at org.apache.camel.processor.SendProcessor.process(SendProcessor.java:148)
at org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:76)
at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:548)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:201)
at org.apache.camel.processor.Pipeline.process(Pipeline.java:138)
at org.apache.camel.processor.Pipeline.process(Pipeline.java:101)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:201)
at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:97)
>> at org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:326)
at org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:215)
有一件事是我还使用了一个kafka主题来发送消息,因为它出现在堆栈跟踪中,所以我不确定是否也需要设置kafka。
1条答案
按热度按时间7lrncoxx1#
在你的路线上,你已经在做一个
.to("direct:agregador")
. 为什么要使用生产者模板将其发送到处理器中?似乎正在发生的是处理器正在发送
someEntity
但当控制权返回到路由时,因为您尚未将exchange的主体设置为someEntity
,则exchange仍保留原始邮件,而不保留idAgregador
标题。您可以在处理器中删除生产者模板代码并添加: