This is an automated email from the ASF dual-hosted git repository. rmaucher pushed a commit to branch 10.1.x in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit d84399bede6644d99f907f12dd6e6d4b087a7073 Author: opencode <[email protected]> AuthorDate: Fri Oct 9 16:18:25 2026 +0200 Release a two phase commit message when its confirmation overtook it Delivery required the confirmation of a stored message to arrive while the original was still in the pending map. On a datagram based transport a confirmation can overtake its original, and the receiver then consumed the confirmation as an error, logged a warning and, when the original arrived moments later, stored it with no further confirmation ever following, so it was silently dropped at expiry. A confirmation whose original is not yet stored is now kept, timestamped, in a map of early confirmations. An original message that arrives while its confirmation is pending there is released immediately. Both handovers are serialised by a small lock so the two paths cannot miss each other, while the notification of the released message happens outside the lock. The heartbeat expires early confirmations whose original never arrives. Add tests for both arrival orders, duplicate confirmations and the expiry of an early confirmation. --- .../group/interceptors/LocalStrings.properties | 1 + .../interceptors/TwoPhaseCommitInterceptor.java | 50 ++++++++++- .../TestTwoPhaseCommitInterceptor.java | 98 ++++++++++++++++++++++ webapps/docs/changelog.xml | 9 ++ 4 files changed, 154 insertions(+), 4 deletions(-) diff --git a/java/org/apache/catalina/tribes/group/interceptors/LocalStrings.properties b/java/org/apache/catalina/tribes/group/interceptors/LocalStrings.properties index da8a19331f..fd9e1d6f9b 100644 --- a/java/org/apache/catalina/tribes/group/interceptors/LocalStrings.properties +++ b/java/org/apache/catalina/tribes/group/interceptors/LocalStrings.properties @@ -120,6 +120,7 @@ throughputInterceptor.report=ThroughputInterceptor Report[\n\ \tRx Speed:{8} MiB/s (since 1st msg)\n\ \tReceived:{9} MiB]\n +twoPhaseCommitInterceptor.expiredConfirmation=Removing confirmation for a message that has not arrived [{0}] twoPhaseCommitInterceptor.expiredMessage=Removing expired message [{0}] twoPhaseCommitInterceptor.heartbeat.failed=Unable to perform heartbeat on the TwoPhaseCommit interceptor. twoPhaseCommitInterceptor.originalMessage.missing=Received a confirmation, but original message is missing. Id:[{0}] diff --git a/java/org/apache/catalina/tribes/group/interceptors/TwoPhaseCommitInterceptor.java b/java/org/apache/catalina/tribes/group/interceptors/TwoPhaseCommitInterceptor.java index eb3cfc4970..b349ba76c3 100644 --- a/java/org/apache/catalina/tribes/group/interceptors/TwoPhaseCommitInterceptor.java +++ b/java/org/apache/catalina/tribes/group/interceptors/TwoPhaseCommitInterceptor.java @@ -34,8 +34,10 @@ import org.apache.juli.logging.LogFactory; /** * Two-phase commit interceptor. The sender transmits a confirmation message after each message. The receiver stores - * each message and only passes it on when the matching confirmation message is received. There is no retransmission, - * so a message whose confirmation is lost, or that arrives out of order, is dropped after the expiration time. + * each message and only passes it on when the matching confirmation message is received. A confirmation that arrives + * before its original message is remembered for the expiration time, so the original is still released when it + * arrives late. There is no retransmission, so a message whose confirmation is lost is dropped after the expiration + * time. */ public class TwoPhaseCommitInterceptor extends ChannelInterceptorBase { @@ -60,6 +62,20 @@ public class TwoPhaseCommitInterceptor extends ChannelInterceptorBase { */ protected final Map<UniqueId,MapEntry> messages = new ConcurrentHashMap<>(); + /** + * Map of confirmations that were received before the original message they confirm, + * mapped to the time of their arrival. An original message that arrives while its + * confirmation is pending here is released immediately instead of waiting for a + * confirmation that has already been consumed. + */ + protected final Map<UniqueId,Long> earlyConfirmations = new ConcurrentHashMap<>(); + + /** + * Lock used to hand over messages and confirmations that arrive concurrently, so that + * the check of one against the other cannot miss an insertion of the other. + */ + private final Object deliveryLock = new Object(); + /** * Message expiration time in milliseconds. */ @@ -113,7 +129,15 @@ public class TwoPhaseCommitInterceptor extends ChannelInterceptorBase { END_DATA, 0, END_DATA.length)) { UniqueId id = new UniqueId(msg.getMessage().getBytesDirect(), START_DATA.length, msg.getUniqueId().length); - MapEntry original = messages.remove(id); + MapEntry original = null; + synchronized (deliveryLock) { + original = messages.remove(id); + if (original == null) { + // The confirmation overtook its original message. Remember the + // confirmation, so the original is released as soon as it arrives + earlyConfirmations.put(id, Long.valueOf(System.currentTimeMillis())); + } + } if (original != null) { super.messageReceived(original.msg); } else { @@ -123,7 +147,16 @@ public class TwoPhaseCommitInterceptor extends ChannelInterceptorBase { } else { UniqueId id = new UniqueId(msg.getUniqueId()); MapEntry entry = new MapEntry((ChannelMessage) msg.deepclone(), id, System.currentTimeMillis()); - messages.put(id, entry); + MapEntry release = null; + synchronized (deliveryLock) { + messages.put(id, entry); + if (earlyConfirmations.remove(id) != null) { + release = messages.remove(id); + } + } + if (release != null) { + super.messageReceived(release.msg); + } } } else { super.messageReceived(msg); @@ -175,6 +208,15 @@ public class TwoPhaseCommitInterceptor extends ChannelInterceptorBase { iter.remove(); } } + Iterator<Map.Entry<UniqueId,Long>> confirmations = earlyConfirmations.entrySet().iterator(); + while (confirmations.hasNext()) { + Map.Entry<UniqueId,Long> entry = confirmations.next(); + if (now - entry.getValue().longValue() > expire) { + log.info(sm.getString("twoPhaseCommitInterceptor.expiredConfirmation", + Arrays.toString(entry.getKey().getBytes()))); + confirmations.remove(); + } + } } catch (Exception e) { log.warn(sm.getString("twoPhaseCommitInterceptor.heartbeat.failed"), e); } finally { diff --git a/test/org/apache/catalina/tribes/group/interceptors/TestTwoPhaseCommitInterceptor.java b/test/org/apache/catalina/tribes/group/interceptors/TestTwoPhaseCommitInterceptor.java index b5ecc22a29..2bff17f132 100644 --- a/test/org/apache/catalina/tribes/group/interceptors/TestTwoPhaseCommitInterceptor.java +++ b/test/org/apache/catalina/tribes/group/interceptors/TestTwoPhaseCommitInterceptor.java @@ -68,6 +68,31 @@ public class TestTwoPhaseCommitInterceptor { return payload; } + // Upstream stand-in: records the messages the interceptor releases + private static class RecordingReceiveChain extends ChannelInterceptorBase { + + private final List<ChannelMessage> messages = Collections.synchronizedList(new ArrayList<>()); + + @Override + public void messageReceived(ChannelMessage msg) { + messages.add(msg); + } + } + + /* + * Runs a message through a sender-side interceptor and returns the confirmation message + * the interceptor generated for it. + */ + private static ChannelMessage buildConfirmation(Member sender, ChannelData original) throws Exception { + TwoPhaseCommitInterceptor senderInterceptor = new TwoPhaseCommitInterceptor(); + RecordingSendChain chain = new RecordingSendChain(); + GroupChannel channel = new GroupChannel(); + senderInterceptor.setChannel(channel); + senderInterceptor.setNext(chain); + senderInterceptor.sendMessage(new Member[] { sender }, original, null); + return chain.messages.get(1); + } + private static ChannelData createMessage(Member source) { ChannelData data = new ChannelData(false); data.setUniqueId(UUIDGenerator.randomUUID(false)); @@ -146,6 +171,79 @@ public class TestTwoPhaseCommitInterceptor { Arrays.equals(savedId, chain.messages.get(1).getUniqueId())); } + @Test + public void testConfirmationArrivingBeforeOriginalReleasesLateOriginal() throws Exception { + MemberImpl member = new MemberImpl("localhost", 4000, -1); + ChannelData original = createMessage(member); + ChannelMessage confirmation = buildConfirmation(member, original); + + TwoPhaseCommitInterceptor receiver = new TwoPhaseCommitInterceptor(); + RecordingReceiveChain released = new RecordingReceiveChain(); + receiver.setPrevious(released); + + receiver.messageReceived(confirmation); + Assert.assertFalse("The confirmation must be held for the original", + receiver.earlyConfirmations.isEmpty()); + Assert.assertTrue("Nothing may be released before the original arrives", + released.messages.isEmpty()); + + receiver.messageReceived(original); + Assert.assertEquals("The late original must be released immediately", + 1, released.messages.size()); + Assert.assertTrue("No bookkeeping may remain", + receiver.messages.isEmpty() && receiver.earlyConfirmations.isEmpty()); + } + + @Test + public void testOriginalThenConfirmationDeliveredOnce() throws Exception { + MemberImpl member = new MemberImpl("localhost", 4000, -1); + ChannelData original = createMessage(member); + ChannelMessage confirmation = buildConfirmation(member, original); + + TwoPhaseCommitInterceptor receiver = new TwoPhaseCommitInterceptor(); + RecordingReceiveChain released = new RecordingReceiveChain(); + receiver.setPrevious(released); + + receiver.messageReceived(original); + Assert.assertTrue("The original must be held for the confirmation", + released.messages.isEmpty()); + + receiver.messageReceived(confirmation); + Assert.assertEquals("The confirmed original must be released", + 1, released.messages.size()); + Assert.assertTrue("No bookkeeping may remain", + receiver.messages.isEmpty() && receiver.earlyConfirmations.isEmpty()); + + // A duplicate confirmation for the same id must not deliver again + receiver.messageReceived(confirmation); + Assert.assertEquals(1, released.messages.size()); + } + + @Test + public void testEarlyConfirmationExpiresWithoutOriginal() throws Exception { + MemberImpl member = new MemberImpl("localhost", 4000, -1); + ChannelData original = createMessage(member); + ChannelMessage confirmation = buildConfirmation(member, original); + + TwoPhaseCommitInterceptor receiver = new TwoPhaseCommitInterceptor(); + receiver.setExpire(50); + RecordingReceiveChain released = new RecordingReceiveChain(); + receiver.setPrevious(released); + + receiver.messageReceived(confirmation); + Assert.assertEquals(1, receiver.earlyConfirmations.size()); + + Thread.sleep(200); + receiver.heartbeat(); + Assert.assertTrue("The stale confirmation must be dropped", + receiver.earlyConfirmations.isEmpty()); + + // The original is no longer matched against the expired confirmation + receiver.messageReceived(original); + Assert.assertTrue("A message after the expired confirmation must not be released", + released.messages.isEmpty()); + } + @Test public void testOriginalSendFailureSkipsConfirmation() throws Exception { TwoPhaseCommitInterceptor interceptor = new TwoPhaseCommitInterceptor(); diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml index 8f00696c7a..c4244eddfe 100644 --- a/webapps/docs/changelog.xml +++ b/webapps/docs/changelog.xml @@ -2542,6 +2542,15 @@ and failures of the original message were never reported at all. (remm) </fix> + <fix> + The <code>TwoPhaseCommitInterceptor</code> now remembers a + confirmation message that arrives before the original message + it confirms, and releases the original as soon as it arrives. + Previously, the confirmation was consumed as an error and the + original message was silently dropped when it arrived and no + further confirmation ever followed, so a single moment of + network reordering cost one lost cluster message. (remm) + </fix> </changelog> </subsection> <subsection name="Other"> --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
