This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 76c81755f256 CAMEL-25855: camel-quickfix - remove the correlation rule
of an InOut request that got no reply (#28378)
76c81755f256 is described below
commit 76c81755f256b9970b0c30990e4847750807ecf3
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 24 19:13:55 2026 +0530
CAMEL-25855: camel-quickfix - remove the correlation rule of an InOut
request that got no reply (#28378)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../component/quickfixj/MessageCorrelator.java | 25 +++-
.../component/quickfixj/QuickfixjProducer.java | 16 ++-
.../QuickfixjProducerCorrelationCleanupTest.java | 130 +++++++++++++++++++++
3 files changed, 166 insertions(+), 5 deletions(-)
diff --git
a/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessageCorrelator.java
b/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessageCorrelator.java
index 9ca073db44e9..8b6a9bcc276d 100644
---
a/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessageCorrelator.java
+++
b/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessageCorrelator.java
@@ -44,14 +44,33 @@ public class MessageCorrelator implements
QuickfixjEventListener {
return new Callable<Message>() {
@Override
public Message call() throws Exception {
- if (!correlationRule.getLatch().await(timeout,
TimeUnit.MILLISECONDS)) {
- throw new
ExchangeTimedOutException(correlationRule.getExchange(), timeout);
+ try {
+ if (!correlationRule.getLatch().await(timeout,
TimeUnit.MILLISECONDS)) {
+ throw new
ExchangeTimedOutException(correlationRule.getExchange(), timeout);
+ }
+ return correlationRule.getReplyMessage();
+ } finally {
+ // a rule that got no reply (timeout, interrupt) must not
stay registered
+ rules.remove(correlationRule);
}
- return correlationRule.getReplyMessage();
}
};
}
+ /**
+ * Removes the correlation rules of the given exchange, for a request that
was not sent so no reply will come.
+ */
+ public void cancelReply(Exchange exchange) {
+ rules.removeIf(rule -> rule.getExchange() == exchange);
+ }
+
+ /**
+ * The number of correlation rules that wait for a reply. Package-private,
for tests.
+ */
+ int pendingReplies() {
+ return rules.size();
+ }
+
@Override
public void onEvent(QuickfixjEventCategory eventCategory, SessionID
sessionID, Message message) throws Exception {
if (message != null) {
diff --git
a/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjProducer.java
b/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjProducer.java
index bf2710b92c16..7392e440c5f2 100644
---
a/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjProducer.java
+++
b/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjProducer.java
@@ -80,13 +80,25 @@ public class QuickfixjProducer extends DefaultProducer {
}
Callable<Message> callable = null;
+ MessageCorrelator messageCorrelator = null;
if (exchange.getPattern().isOutCapable()) {
- MessageCorrelator messageCorrelator =
getEndpoint().getEngine().getMessageCorrelator();
+ messageCorrelator =
getEndpoint().getEngine().getMessageCorrelator();
+ // registered before the send so that a fast reply is not missed
callable = messageCorrelator.getReply(exchange);
}
- if (!session.send(message)) {
+ boolean sent = false;
+ try {
+ sent = session.send(message);
+ } finally {
+ if (!sent && messageCorrelator != null) {
+ // the exchange fails, so nobody waits for the reply
(QuickFIX/J has stored an application message before
+ // it found the session not logged on, and may still resend it
after the next logon)
+ messageCorrelator.cancelReply(exchange);
+ }
+ }
+ if (!sent) {
throw new CannotSendException("Cannot send FIX message: " +
message.toString());
}
diff --git
a/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjProducerCorrelationCleanupTest.java
b/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjProducerCorrelationCleanupTest.java
new file mode 100644
index 000000000000..16a4b1915b59
--- /dev/null
+++
b/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjProducerCorrelationCleanupTest.java
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.quickfixj;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.ExchangeTimedOutException;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultExchange;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+import quickfix.FixVersions;
+import quickfix.Message;
+import quickfix.MessageUtils;
+import quickfix.Session;
+import quickfix.SessionID;
+import quickfix.field.BeginString;
+import quickfix.field.MsgType;
+import quickfix.field.SenderCompID;
+import quickfix.field.TargetCompID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.mockito.ArgumentMatchers.isA;
+
+/**
+ * An InOut exchange that gets no reply (timeout) or whose message cannot be
sent must not leave its correlation rule
+ * registered in the engine's {@link MessageCorrelator}.
+ */
+class QuickfixjProducerCorrelationCleanupTest {
+
+ private CamelContext context;
+ private QuickfixjEngine engine;
+ private QuickfixjProducer producer;
+ private Session session;
+ private Message fixMessage;
+ private SessionID responseSessionID;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ context = new DefaultCamelContext();
+ context.start();
+ engine = TestSupport.createEngine();
+
+ fixMessage = new Message();
+ fixMessage.getHeader().setString(BeginString.FIELD,
FixVersions.BEGINSTRING_FIX44);
+ fixMessage.getHeader().setString(SenderCompID.FIELD, "SENDER");
+ fixMessage.getHeader().setString(TargetCompID.FIELD, "TARGET");
+ SessionID sessionID = MessageUtils.getSessionID(fixMessage);
+ responseSessionID = new SessionID(sessionID.getBeginString(),
sessionID.getTargetCompID(), sessionID.getSenderCompID());
+
+ QuickfixjEndpoint endpoint = Mockito.mock(QuickfixjEndpoint.class);
+ Mockito.when(endpoint.getEngine()).thenReturn(engine);
+ Mockito.when(endpoint.getSID()).thenReturn(sessionID);
+
+ session = Mockito.spy(TestSupport.createSession(sessionID));
+ producer = Mockito.spy(new QuickfixjProducer(endpoint));
+ Mockito.doReturn(session).when(producer).getSession(sessionID);
+ }
+
+ @AfterEach
+ void tearDown() {
+ context.stop();
+ }
+
+ @Test
+ void timedOutRequestsDoNotStayRegistered() throws Exception {
+ Mockito.doReturn(true).when(session).send(isA(Message.class));
+
+ for (int i = 0; i < 50; i++) {
+ Exchange exchange = createInOutExchange(10L);
+ producer.process(exchange);
+ assertInstanceOf(ExchangeTimedOutException.class,
exchange.getException());
+ }
+
+ assertEquals(0, pendingRules(), "correlation rules left after 50 timed
out requests");
+ }
+
+ @Test
+ void requestThatCannotBeSentDoesNotStayRegistered() throws Exception {
+ Mockito.doReturn(false).when(session).send(isA(Message.class));
+
+ Exchange exchange = createInOutExchange(5000L);
+ producer.process(exchange);
+ assertInstanceOf(CannotSendException.class, exchange.getException());
+
+ assertEquals(0, pendingRules(), "correlation rules left after a failed
send");
+ }
+
+ @Test
+ void requestWhoseSendThrowsDoesNotStayRegistered() throws Exception {
+ Mockito.doThrow(new IllegalStateException("not logged
on")).when(session).send(isA(Message.class));
+
+ Exchange exchange = createInOutExchange(5000L);
+ producer.process(exchange);
+ assertInstanceOf(IllegalStateException.class, exchange.getException());
+
+ assertEquals(0, pendingRules(), "correlation rules left after a send
that threw");
+ }
+
+ private Exchange createInOutExchange(long timeout) {
+ Exchange exchange = new DefaultExchange(context,
ExchangePattern.InOut);
+ exchange.getIn().setBody(fixMessage);
+ exchange.setProperty(QuickfixjProducer.CORRELATION_CRITERIA_KEY,
+ new MessagePredicate(responseSessionID,
MsgType.EXECUTION_REPORT));
+ exchange.setProperty(QuickfixjProducer.CORRELATION_TIMEOUT_KEY,
timeout);
+ return exchange;
+ }
+
+ private int pendingRules() {
+ return engine.getMessageCorrelator().pendingReplies();
+ }
+}
