Apache Camel: respuesta recibida para correlationID desconocido
Hay unmiddleware
en medio de otros dos componentes de software. En elmiddleware
Estoy encaminandoApache ActiveMQ
mensajes porApache Camel
.
Así es como funciona:
1stComponent
usosmiddleware
enviar mensaje a la3rdComponent
3rdComponent
responde el mensaje y lo envía de nuevo a la1st
(utilizandomiddleware
).
1stComponent <<=>> Middleware <<=>> 3rdComponent
Problema:
Estoy usandoConcurrentConsumers
en middleware.
En medio de enviar muchos mensajes de forma secuencial, de repentemiddleware
Se detiene todo el proceso! No hay excepciones ni mensajes! por ejemplo, primero se procesaron 100 de 500 mensajes y los restos permanecen en la cola como mensajes pendientes.
esta advertencia se registra a veces en medio del proceso:
[WARN ] TemporaryQueueReplyManager(Camel (camel-1) thread #11 - TemporaryQueueReplyManager[Q.MyQ]):91 - Reply received for unknown correlationID [c551c7aa061f501c]. The message will be ignored: ActiveMQMapMessage {commandId = 2161, responseRequired = true, messageId = ID:xxxxxxx, originalDestination = null, originalTransactionId = null, producerId = ID:xxxxxxx, destination = temp-queue://ID:localhost.localdomain-40961-1389890357282-3:1:1, transactionId = null, expiration = 0, timestamp = 1389890272360, arrival = 0, brokerInTime = 1389890272360, brokerOutTime = 1389890272360, correlationId = c551c7aa061f501c, replyTo = temp-queue://ID:localhost.localdomain-40961-1389890357282-3:1:1, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = org.apache.activemq.util.ByteSequence@19e19da, dataStructure = null, redeliveryCounter = 0, size = 0, properties = {breadcrumbId=ID:xxxxxxxxxxxxxx, Title=300, CamelJmsDeliveryMode=1}, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false} ActiveMQMapMessage{ theTable = {} }
este es elmiddlewares
Código:
private static class MyRouteBuilder extends RouteBuilder {
@Override
public void configure() throws Exception {
from("activemq:queue:Q.Middleware?concurrentConsumers=1&maxConcurrentConsumers=10")
.threads(1, 100)
.process(new Processor() {
public void process(Exchange exchange) {
//some code
}
})
.inOut("activemq2:queue:Q.3RD")
;
}
}
y el3rdComponent
:
private static class MyRouteBuilder extends RouteBuilder {
@Override
public void configure() {
from("activemq:queue:Q.3RD")
.threads(1, 100)
.process(new Processor() {
public void process(Exchange exchange) {
//some code
}
})
;
}
}