This is an automated email from the ASF dual-hosted git repository. rmaucher pushed a commit to branch 9.0.x in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 1a3a1f4843a592e8137a13f924657164c6cc6b1d Author: opencode <[email protected]> AuthorDate: Fri Oct 9 16:04:52 2026 +0200 Report two phase commit send results through the application error handler The interceptor passed a null payload when sending the original message and the payload when sending its internal confirmation message. The payload carries the ErrorHandler the application supplied to Channel.send and the only consumer of it, the MessageDispatchInterceptor, invokes it with the id of the message it actually transmits. With the MessageDispatchInterceptor performing the send, the application therefore received completion and error callbacks for the confirmation's random id, which it cannot correlate with the id returned by the send, and a failure of the original message was never reported to the handler at all. Forward the payload with the original message, as every other interceptor does, and send the confirmation without a payload, so that exactly one completion or error callback is issued, correlated with the id of the message the application sent. Add a unit test covering the payload routing, the null passthrough and the send failure path. --- .../interceptors/TwoPhaseCommitInterceptor.java | 8 +- .../TestTwoPhaseCommitInterceptor.java | 149 +++++++++++++++++++++ webapps/docs/changelog.xml | 12 ++ 3 files changed, 167 insertions(+), 2 deletions(-) diff --git a/java/org/apache/catalina/tribes/group/interceptors/TwoPhaseCommitInterceptor.java b/java/org/apache/catalina/tribes/group/interceptors/TwoPhaseCommitInterceptor.java index 3f0d7d1fbd..eb3cfc4970 100644 --- a/java/org/apache/catalina/tribes/group/interceptors/TwoPhaseCommitInterceptor.java +++ b/java/org/apache/catalina/tribes/group/interceptors/TwoPhaseCommitInterceptor.java @@ -78,7 +78,11 @@ public class TwoPhaseCommitInterceptor extends ChannelInterceptorBase { // msg.setOptions(msg.getOptions() & (~getOptionFlag()) // and just send one message if (okToProcess(msg.getOptions())) { - super.sendMessage(destination, msg, null); + // The payload carries the application error handler. It applies to the real + // message, so forward it with the original. The confirmation is an internal + // message with a random id the application cannot correlate, so it goes without + // a payload, which also keeps the handler from being notified twice per send + super.sendMessage(destination, msg, payload); ChannelMessage confirmation; if (deepclone) { confirmation = (ChannelMessage) msg.deepclone(); @@ -90,7 +94,7 @@ public class TwoPhaseCommitInterceptor extends ChannelInterceptorBase { confirmation.getMessage().append(START_DATA, 0, START_DATA.length); confirmation.getMessage().append(msg.getUniqueId(), 0, msg.getUniqueId().length); confirmation.getMessage().append(END_DATA, 0, END_DATA.length); - super.sendMessage(destination, confirmation, payload); + super.sendMessage(destination, confirmation, null); } else { // turn off two phase commit // this won't work if the interceptor has 0 as a flag diff --git a/test/org/apache/catalina/tribes/group/interceptors/TestTwoPhaseCommitInterceptor.java b/test/org/apache/catalina/tribes/group/interceptors/TestTwoPhaseCommitInterceptor.java new file mode 100644 index 0000000000..7f33f2ce2d --- /dev/null +++ b/test/org/apache/catalina/tribes/group/interceptors/TestTwoPhaseCommitInterceptor.java @@ -0,0 +1,149 @@ +/* + * 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.catalina.tribes.group.interceptors; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import org.junit.Assert; +import org.junit.Test; + +import org.apache.catalina.tribes.ChannelException; +import org.apache.catalina.tribes.ChannelMessage; +import org.apache.catalina.tribes.ErrorHandler; +import org.apache.catalina.tribes.Member; +import org.apache.catalina.tribes.UniqueId; +import org.apache.catalina.tribes.group.ChannelInterceptorBase; +import org.apache.catalina.tribes.group.GroupChannel; +import org.apache.catalina.tribes.group.InterceptorPayload; +import org.apache.catalina.tribes.io.ChannelData; +import org.apache.catalina.tribes.io.XByteBuffer; +import org.apache.catalina.tribes.membership.MemberImpl; +import org.apache.catalina.tribes.util.UUIDGenerator; + +public class TestTwoPhaseCommitInterceptor { + + private static class RecordingSendChain extends ChannelInterceptorBase { + + private final List<ChannelMessage> messages = Collections.synchronizedList(new ArrayList<>()); + private final List<InterceptorPayload> payloads = Collections.synchronizedList(new ArrayList<>()); + + @Override + public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) + throws ChannelException { + messages.add(msg); + payloads.add(payload); + } + } + + private static InterceptorPayload createPayload() { + InterceptorPayload payload = new InterceptorPayload(); + payload.setErrorHandler(new ErrorHandler() { + @Override + public void handleError(ChannelException x, UniqueId id) { + // Not used by these tests + } + + @Override + public void handleCompletion(UniqueId id) { + // Not used by these tests + } + }); + return payload; + } + + private static ChannelData createMessage(Member source) { + ChannelData data = new ChannelData(false); + data.setUniqueId(UUIDGenerator.randomUUID(false)); + data.setAddress(source); + XByteBuffer buffer = new XByteBuffer(64, false); + byte[] content = "payload-data".getBytes(); + buffer.append(content, 0, content.length); + data.setMessage(buffer); + return data; + } + + @Test + public void testPayloadAppliesToOriginalMessageOnly() throws Exception { + TwoPhaseCommitInterceptor interceptor = new TwoPhaseCommitInterceptor(); + RecordingSendChain chain = new RecordingSendChain(); + GroupChannel channel = new GroupChannel(); + interceptor.setChannel(channel); + interceptor.setNext(chain); + + MemberImpl member = new MemberImpl("localhost", 4000, -1); + InterceptorPayload payload = createPayload(); + ChannelData msg = createMessage(member); + byte[] originalId = msg.getUniqueId(); + + interceptor.sendMessage(new Member[] { member }, msg, payload); + + Assert.assertEquals("The original message and the confirmation must both be sent", + 2, chain.messages.size()); + Assert.assertSame("The original message must be sent first", msg, chain.messages.get(0)); + Assert.assertSame("The application payload must be forwarded with the original message", + payload, chain.payloads.get(0)); + Assert.assertNull("The confirmation must be sent without the application payload", + chain.payloads.get(1)); + Assert.assertFalse("The confirmation must carry its own id", + Arrays.equals(originalId, chain.messages.get(1).getUniqueId())); + } + + @Test + public void testNullPayloadStaysNull() throws Exception { + TwoPhaseCommitInterceptor interceptor = new TwoPhaseCommitInterceptor(); + RecordingSendChain chain = new RecordingSendChain(); + GroupChannel channel = new GroupChannel(); + interceptor.setChannel(channel); + interceptor.setNext(chain); + + MemberImpl member = new MemberImpl("localhost", 4000, -1); + interceptor.sendMessage(new Member[] { member }, createMessage(member), null); + + Assert.assertEquals(2, chain.messages.size()); + Assert.assertNull(chain.payloads.get(0)); + Assert.assertNull(chain.payloads.get(1)); + } + + @Test + public void testOriginalSendFailureSkipsConfirmation() throws Exception { + TwoPhaseCommitInterceptor interceptor = new TwoPhaseCommitInterceptor(); + RecordingSendChain chain = new RecordingSendChain() { + @Override + public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) + throws ChannelException { + throw new ChannelException("Induced failure"); + } + }; + GroupChannel channel = new GroupChannel(); + interceptor.setChannel(channel); + interceptor.setNext(chain); + + MemberImpl member = new MemberImpl("localhost", 4000, -1); + try { + interceptor.sendMessage(new Member[] { member }, createMessage(member), createPayload()); + Assert.fail("The send must fail"); + } catch (ChannelException expected) { + // Expected: the failure of the original send reaches the caller and the + // error handler of the payload that was forwarded for the original + } + Assert.assertEquals("No confirmation may be sent after the original failed", + 0, chain.messages.size()); + } +} diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml index 49995bb8e0..b2a1c1ec89 100644 --- a/webapps/docs/changelog.xml +++ b/webapps/docs/changelog.xml @@ -2496,6 +2496,18 @@ with membership listeners that call back into the interceptor. (remm) </fix> + <fix> + The <code>TwoPhaseCommitInterceptor</code> now forwards the + interceptor payload, and with it the <code>ErrorHandler</code> + supplied by the application, with the original message instead + of with its internal confirmation message. When the + <code>MessageDispatchInterceptor</code> performed the actual + send, completion and error callbacks were previously reported + under the confirmation message's random id, which the + application cannot correlate with the id returned by the send, + and failures of the original message were never reported at + all. (remm) + </fix> </changelog> </subsection> <subsection name="Other"> --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
