Correlation Identifier no Mule: o padrão do livro e o correlationId

No Mule 4, “correlation ID” tem dois sentidos. Um é o padrão Correlation Identifier do livro Enterprise Integration Patterns, que liga uma resposta à requisição que a originou. O outro é o correlationId do Mule, que acompanha uma requisição de uma aplicação para outra. Esta página mostra os dois no cenário da série, um pedido cobrado que o provedor de pagamento confirma depois, com o código e as suítes MUnit do repositório mulesoft-patterns-examples.

O provedor responde depois: de qual requisição é este callback?

O livro parte de uma aplicação que fez uma requisição por mensagens e recebeu uma resposta, e pergunta como ela sabe a que requisição essa resposta pertence (Correlation Identifier).

No exemplo, depois de cobrar um pedido, o consumidor de pagamento pede o pagamento a um provedor, que aceita com 202 e, mais tarde, chama um endpoint de callback para dizer se aprovou ou recusou. Outros pedidos foram enviados nesse meio-tempo, e os callbacks podem chegar em qualquer ordem. O consumidor precisa descobrir de qual pedido é cada um.

O padrão no livro: o ID da requisição volta na resposta

A solução do livro é que cada resposta traga um correlation identifier, um ID único que indica a que requisição ela responde. O padrão tem seis partes:

• o requestor, a aplicação que envia a requisição e espera a resposta;
• o replier, a aplicação que recebe a requisição, atende e responde;
• a requisição;
• a resposta;
• o request ID, que vai na requisição e a identifica de forma única;
• o correlation ID, que vai na resposta com o mesmo valor do request ID, copiado pelo replier.

Diagrama: o padrão no livro

O diagrama responde como o requestor liga cada resposta à sua requisição quando as respostas voltam fora de ordem.

O requestor guarda as requisições que esperam resposta e procura o correlation ID de cada resposta entre elas. A ordem de chegada não serve de pista: nada garante que a primeira resposta seja a da primeira requisição.

O padrão do livro no Mule: o callback do provedor de pagamento

O Mule não tem um componente para esse padrão. No exemplo, ele é montado com HTTP Request, HTTP Listener e dois object stores, em correlation-identifier.xml. Depois da cobrança do idempotent receiver, o sub-flow payment-request-from-provider faz o papel do requestor. O replier é o flow payment-provider-simulator, um provedor simulado na própria aplicação.

Diagrama: o pagamento confirmado por callback

O diagrama responde por onde passa o requestId, da cobrança até o resultado gravado no pedido, e o que acontece com um callback que nenhuma requisição espera.

A requisição carrega o requestId, não o orderId

O sub-flow gera um requestId ("PRQ-" ++ uuid()) e envia ao provedor só requestId e amount: o provedor não tem por que conhecer os pedidos da aplicação. Antes da chamada, ele guarda a requisição inteira, com o orderId e o amount, no store de pendentes, com o requestId como chave, porque o callback pode chegar antes do 202.

<!-- Pending requests by requestId. TTL 1 hour: a callback later than that is treated as unknown. -->
<os:object-store name="eip-pending-payment-requests-store" doc:name="Pending payment requests"
	persistent="false" entryTtl="1" entryTtlUnit="HOURS" expirationInterval="5" expirationIntervalUnit="MINUTES" />

No repositório: src/main/mule/eip/correlation-identifier/correlation-identifier.xml

<!-- Stored before the call: the callback can arrive before the provider's 202 does. -->
<os:store doc:name="Remember pending request" key="#[vars.paymentRequest.requestId]" objectStore="eip-pending-payment-requests-store">
	<os:value><![CDATA[#[vars.paymentRequest]]]></os:value>
</os:store>
<os:store doc:name="Order is waiting for payment" key="#[vars.paymentRequest.orderId]" objectStore="eip-order-payment-status-store">
	<os:value><![CDATA[#[output application/java --- { orderId: vars.paymentRequest.orderId, status: "PAYMENT_REQUESTED", requestId: vars.paymentRequest.requestId }]]]></os:value>
</os:store>
<!-- The order is kept aside and put back after the call, so the caller still has it as payload. -->
<set-variable doc:name="Keep the order" variableName="chargedOrder" value="#[payload]" />
<http:request doc:name="Send payment request to provider" config-ref="eip-self-http-request-config" method="POST" path="/eip/provider/payments">
	<http:body><![CDATA[#[output application/json --- vars.providerRequestBody]]]></http:body>
</http:request>
<set-payload doc:name="Order back as payload" value="#[vars.chargedOrder]" />

O pedido vai para uma variável e volta como payload depois da chamada, porque o flow de cobrança continua usando o pedido.

O provedor simulado responde 202 e chama de volta

O provedor responde 202 e, dentro de um escopo async, chama o caminho configurado em eip.provider.callbackPath, como um provedor real chama o webhook cadastrado nele. O flow principal não espera o async terminar (Async), então o 202 não depende do callback, que devolve o requestId no campo correlationId:

<set-variable doc:name="Provider request" variableName="providerRequest" value="#[payload]" />
<async doc:name="Approve later">
	<http:request doc:name="Call back the requestor" config-ref="eip-self-http-request-config" method="POST" path="${eip.provider.callbackPath}">
		<http:body><![CDATA[#[output application/json --- {
	correlationId: vars.providerRequest.requestId,
	status: "APPROVED",
	approvedAmount: vars.providerRequest.amount,
	providerReference: "PRV-" ++ uuid()
}]]]></http:body>
	</http:request>
</async>
<set-payload doc:name="Accepted" value="#[output application/json --- { requestId: vars.providerRequest.requestId, status: 'ACCEPTED' }]" />

No repositório: src/main/mule/eip/correlation-identifier/correlation-identifier.xml

O provedor simulado sempre aprova o valor pedido. Os outros resultados aparecem nos testes que enviam o callback eles mesmos.

O callback reivindica a requisição pendente

O flow payment-confirm-from-provider-callback usa só o correlationId do corpo. Sem correlationId, responde 400. Com ele, lê a requisição pendente e a remove do store antes de mexer no pedido: é assim que o callback reivindica a requisição. Se a chave não existe, o os:retrieve ou o os:remove lança OS:KEY_NOT_FOUND (referência do Object Store), e o escopo try transforma esse erro num 404. O motivo pode ser um ID desconhecido, uma entrada expirada, um callback já ligado antes ou outra cópia que removeu a requisição entre o os:retrieve e o os:remove desta.

<set-variable doc:name="Callback" variableName="callback" value="#[payload]" />
<set-variable doc:name="Not matched yet" variableName="unmatched" value="#[false]" />
<try doc:name="Match and claim the pending request">
	<os:retrieve doc:name="Find pending request" key="#[vars.callback.correlationId]" objectStore="eip-pending-payment-requests-store" target="pendingRequest" />
	<os:remove doc:name="Claim the pending request" key="#[vars.callback.correlationId]" objectStore="eip-pending-payment-requests-store" />
	<error-handler>
		<!-- Unknown, expired, already matched, or claimed by another copy in between: all the same 404. -->
		<on-error-continue doc:name="No pending request" type="OS:KEY_NOT_FOUND" logException="false">
			<set-variable doc:name="Not matched" variableName="unmatched" value="#[true]" />
		</on-error-continue>
	</error-handler>
</try>
<choice doc:name="Matched?">
	<when expression="#[vars.unmatched default false]">
		<logger doc:name="Log unmatched callback" level="WARN" message="#['Unmatched provider callback, correlationId in body=' ++ vars.callback.correlationId ++ ', correlationId=' ++ correlationId]" />
		<raise-error doc:name="Unknown correlationId" type="APP:NOT_FOUND" description="No pending payment request for this correlationId" />
	</when>
</choice>

No repositório: src/main/mule/eip/correlation-identifier/correlation-identifier.xml

Com a requisição em mãos, o flow decide o resultado pelo status do callback e pelo valor aprovado:

<choice doc:name="Route on callback status">
	<when expression="#[vars.callback.status == 'APPROVED' and vars.callback.approvedAmount == vars.pendingRequest.amount]">
		<set-variable doc:name="Paid" variableName="outcome" value="#[{ status: 'PAID' }]" />
	</when>
	<when expression="#[vars.callback.status == 'DECLINED']">
		<set-variable doc:name="Declined" variableName="outcome" value="#[{ status: 'PAYMENT_DECLINED' }]" />
	</when>
	<when expression="#[vars.callback.status == 'APPROVED']">
		<set-variable doc:name="Review: amount mismatch" variableName="outcome" value="#[{ status: 'PAYMENT_REVIEW', reviewReason: 'AMOUNT_MISMATCH' }]" />
	</when>
	<otherwise>
		<set-variable doc:name="Review: unknown status" variableName="outcome" value="#[{ status: 'PAYMENT_REVIEW', reviewReason: 'UNKNOWN_STATUS' }]" />
	</otherwise>
</choice>

O flow grava no store de status o resultado, o motivo da revisão quando há um, o requestId, o valor pedido (requestedAmount) e, do provedor, o status (providerStatus), o approvedAmount e o providerReference. O status pode ser consultado em GET /eip/payments/{orderId}/status. Os três resultados respondem 200: o provedor já deu a resposta final, e recusar o callback só o faria enviar a mesma resposta de novo.

Callback	Resposta	Pedido	Teste
APPROVED com o valor pedido	200 { result: MATCHED, orderStatus: PAID }	PAID	cid-order-is-paid-by-the-provider-callback, cid-out-of-order-callbacks-are-matched-to-their-orders
DECLINED	200, orderStatus: PAYMENT_DECLINED	PAYMENT_DECLINED	cid-declined-callback-marks-the-order-payment-declined
APPROVED com outro valor (90 em vez de 100)	200, orderStatus: PAYMENT_REVIEW	PAYMENT_REVIEW, reviewReason: AMOUNT_MISMATCH	cid-approved-callback-for-another-amount-goes-to-payment-review
Outro status (CHARGEBACK)	200, orderStatus: PAYMENT_REVIEW	PAYMENT_REVIEW, reviewReason: UNKNOWN_STATUS	cid-callback-with-unknown-status-goes-to-payment-review
Repetição de um callback já ligado, ou um DECLINED atrasado	404	Fica com o primeiro resultado	cid-repeated-callback-returns-404-and-keeps-the-first-outcome
Cópia que perde a requisição entre o os:retrieve e o os:remove	404	Não muda	cid-callback-that-loses-the-claim-returns-404-and-changes-nothing
correlationId desconhecido	404	Nenhum pedido	cid-unmatched-callback-returns-404
Sem correlationId	400	Nenhum pedido	cid-callback-without-correlation-id-returns-400

No teste da cópia que perde a requisição, um mock faz o os:remove lançar OS:KEY_NOT_FOUND, como se outra cópia tivesse removido a chave antes; o pedido continua em PAYMENT_REQUESTED. O exemplo não verifica quem enviou o callback. Num webhook real, verifique também a assinatura do provedor antes de confiar no corpo.

Como tudo roda na mesma aplicação, o header X-CORRELATION-ID do Mule também chega com o callback. Um provedor real não o devolve, e o flow não o lê: a ligação depende só do correlationId no corpo.

Callbacks fora de ordem, nos testes

A suíte correlation-identifier-test-suite.xml tem 14 testes, que passam no Mule 4.9.17 com o MUnit 3.7.1. Além dos que estão na tabela acima:

• cid-out-of-order-callbacks-are-matched-to-their-orders. Pede o pagamento de ORD-CID-A (100) e de ORD-CID-B (250) com o provedor num mock que nunca chama de volta, e o próprio teste envia os callbacks, B antes de A. Cada pedido fica PAID com o seu providerReference e o seu valor, e as duas requisições saem dos pendentes.
• cid-order-is-paid-by-the-provider-callback. De ponta a ponta: o provedor simulado chama de volta sozinho, do async, e o teste lê o store de status com os:retrieve, num until-successful, até o pedido ORD-CID-E2E ficar PAID, com valor 42.
• cid-request-carries-its-id-and-not-the-order. O corpo enviado ao provedor tem só as chaves requestId e amount, e a requisição pendente guarda o orderId.

O correlationId do Mule: o mesmo ID nos logs das duas aplicações

Todo evento do Mule tem um correlationId, que liga as linhas de log a uma execução. O Mule primeiro procura um na origem (por exemplo, numa mensagem JMS ou, no HTTP Listener, no header X-CORRELATION-ID) e só gera um novo quando ele não vem (Correlation ID). O HTTP Request tem a opção Send Correlation Id, com AUTO, ALWAYS e NEVER, e AUTO como padrão (HTTP Connector).

No Mule 4.9.17, o teste mule-cid-http-request-sends-the-caller-correlation-id mostra que, com a opção no padrão, o flow chamado recebe em X-CORRELATION-ID o ID de quem chamou e o adota. Então, quando uma aplicação A chama uma aplicação B com a opção no padrão, a B adota o ID da A, porque também é uma aplicação Mule e recebe a chamada num HTTP Listener. Uma busca por esse ID nos logs traz a requisição nas duas aplicações.

Diagrama: o X-CORRELATION-ID entre duas aplicações

O diagrama responde de onde vem o correlationId da aplicação chamada, com a opção no padrão e com NEVER.

Em mule-correlation-id.xml, o flow chamado devolve o header que recebeu e o correlationId do seu evento:

<flow name="mule-correlation-id-echo">
	<http:listener doc:name="GET /eip/correlation-id/echo" config-ref="http-listener-config" path="/eip/correlation-id/echo" allowedMethods="GET" />
	<ee:transform doc:name="What arrived">
		<ee:message>
			<ee:set-payload><![CDATA[%dw 2.0
output application/json
---
{ receivedHeader: attributes.headers["x-correlation-id"], calleeCorrelationId: correlationId }]]></ee:set-payload>
		</ee:message>
	</ee:transform>
</flow>

No repositório: src/main/mule/eip/correlation-identifier/mule-correlation-id.xml

Os flows de /eip/correlation-id/call/default e /call/never o chamam, e as duas chamadas diferem só em sendCorrelationId:

<http:request doc:name="Call echo (default)" config-ref="eip-self-http-request-config" method="GET" path="/eip/correlation-id/echo" />
<http:request doc:name="Call echo (NEVER)" config-ref="eip-self-http-request-config" method="GET" path="/eip/correlation-id/echo" sendCorrelationId="NEVER" />

O teste mule-cid-send-correlation-id-never-does-not-send-it confirma que, com NEVER, o header não chega e o flow chamado ganha outro ID. As chamadas dos dois testes são HTTP de verdade, mas dentro da mesma aplicação: o HTTP Request aponta para localhost, na porta dela. Entre duas aplicações o mecanismo é o mesmo header; o teste só não passa pela rede entre duas aplicações implantadas.

Para trocar o ID num trecho do flow, a documentação indica o escopo with-correlation-id do módulo Mule Tracing (Correlation ID).

Os dois sentidos lado a lado

	Correlation Identifier (livro)	correlationId (Mule)
Para que serve	Ligar uma resposta à sua requisição	Seguir uma execução nos logs, entre aplicações
Quem gera	O requestor; o replier só copia	O Mule, quando a origem não traz um
Por onde viaja	Na requisição e na resposta (no exemplo, no corpo JSON)	No HTTP, no header X-CORRELATION-ID; em outras origens, no campo de correlação delas
Que pergunta responde	De qual requisição é esta resposta?	Por onde passou esta requisição?

Cuidados com cada um dos IDs

O correlationId do Mule não identifica uma mensagem de negócio. As chamadas que saem da mesma execução carregam o mesmo ID, e um cliente que reenvia sem o header chega com um ID novo. Por isso ele não serve como chave de idempotência, e é justamente o ID que o Idempotent Message Validator usa por padrão; a página do idempotent receiver mostra os dois erros em testes.

A requisição pendente expira. O store de pendentes tem TTL de 1 hora (entryTtl) e verifica as entradas vencidas a cada 5 minutos (expirationInterval). Pela referência do Object Store, quem remove as entradas é a thread de expiração, que roda nesse intervalo; então a entrada sai na primeira verificação depois de vencida, não exatamente ao completar 1 hora. Depois que ela sai, o callback recebe 404, como um ID desconhecido, e o pedido continua em PAYMENT_REQUESTED, porque o store de status não tem TTL. Nenhum teste cobre a expiração: isso vem da configuração e do código.

Reivindicar antes de atualizar estreita a corrida, mas não a fecha. O flow remove a requisição pendente antes de gravar o resultado no pedido, e os testes mostram as duas consequências esperadas: a repetição de um callback recebe 404 e o pedido fica com o primeiro resultado; uma cópia que perde a requisição entre o os:retrieve e o os:remove recebe 404 e não muda nada. Ficam dois limites, que vêm do código e não têm teste:

• O os:retrieve e o os:remove são duas chamadas, não um lock. Se a segunda de duas cópias simultâneas falha no os:remove depende do store. Com stores persistentes no CloudHub, que usam o Object Store v2, não há distributed locking, e a documentação recomenda um lock externo quando vários componentes atualizam o mesmo store ao mesmo tempo (Object Store v2). Então, com stores persistentes, duas cópias tratadas no mesmo instante, por dois workers, ainda podem atualizar o pedido as duas. Para impedir isso de vez, é preciso uma reivindicação atômica ou um lock fora deste flow.
• Se a gravação no pedido falha depois da reivindicação, a requisição já saiu dos pendentes: a nova tentativa do provedor recebe 404, e o pedido fica em PAYMENT_REQUESTED até alguém verificar.

Os stores estão em memória (persistent="false"), para cada execução começar limpa. Um restart apaga as requisições pendentes, e os callbacks delas recebem 404; isso vem do código, sem teste. Um store em memória também não é compartilhado entre réplicas: com dois workers fora de um cluster do Mule, o callback pode chegar ao worker que não guardou a requisição e receber 404. Num cluster de alta disponibilidade do Mule, os object stores, persistentes ou não, ficam numa grade de memória compartilhada entre os nós (clusters de HA). Fora de um cluster, o README do repositório indica um store persistente e compartilhado, como o Object Store v2 no CloudHub, com o limite de lock do item anterior.

Quando o padrão do livro sobra: a resposta na mesma chamada HTTP

O livro descreve duas formas de receber a resposta (Request-Reply). No bloqueio síncrono, o requestor envia e espera, o que pressupõe uma única requisição pendente ou um canal de resposta exclusivo. No callback assíncrono, várias requisições pendentes dividem o canal de resposta, e é aí que o Correlation Identifier entra.

Uma chamada HTTP síncrona está no primeiro caso: o HTTP Request espera a resposta na mesma conexão, já ligada à requisição. Se o provedor devolvesse a aprovação no corpo da própria chamada, um ID de requisição guardado num store só acrescentaria peças. O padrão passa a ser necessário quando a resposta chega por outro caminho: um callback, uma fila de resposta, um arquivo de retorno.

Rodar o exemplo e os testes

Requisitos: Java 17, Maven 3.9 ou mais novo e as credenciais do repositório Enterprise da MuleSoft no ~/.m2/settings.xml, porque o MUnit roda no runtime Enterprise.

git clone https://github.com/brunosouzas/mulesoft-patterns-examples.git
cd mulesoft-patterns-examples
mvn clean test

Para chamar os endpoints, rode a aplicação no Anypoint Studio (porta 8081) e abra o requests.http no REST Client do VS Code ou no IntelliJ:

1. POST /eip/payments/after/charges com { "orderId": "ORD-2001", "amount": 100 } cobra o pedido e pede o pagamento ao provedor simulado.
2. GET /eip/payments/ORD-2001/status mostra o resultado gravado pelo callback: com o provedor simulado, PAID, com requestId, requestedAmount, providerStatus, approvedAmount e providerReference.
3. POST /eip/payments/callback com correlationId PRQ-UNKNOWN recebe 404.
4. GET /eip/correlation-id/call/default e /call/never devolvem o ID de quem chamou ao lado do header recebido e do ID do flow chamado.

Os outros padrões da série, com o componente do Mule de cada um, estão no catálogo de padrões de integração no Mule.