This is an automated email from the ASF dual-hosted git repository.
markt-asf pushed a commit to branch 9.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
The following commit(s) were added to refs/heads/9.0.x by this push:
new d580193e6f Review "Do not poll or swallow interrupts when waiting for
a pooled ..."
d580193e6f is described below
commit d580193e6f17bd43fc910d37a906c18fd5562d2e
Author: Mark Thomas <[email protected]>
AuthorDate: Fri Oct 23 18:47:51 2026 +0100
Review "Do not poll or swallow interrupts when waiting for a pooled ..."
Expand to other locations where interrupts were incorrectly swallowed
Co-authored by: Claude / Sonnet 5.5
Co-authored by: CoPilot / GPT-5.6 Sol
---
.../apache/catalina/tribes/group/GroupChannel.java | 5 +-
.../group/interceptors/NonBlockingCoordinator.java | 139 +++++++++++----------
.../tribes/membership/McastServiceImpl.java | 6 +-
.../membership/StaticMembershipProvider.java | 2 +-
.../catalina/tribes/transport/PooledSender.java | 4 +-
.../tribes/transport/nio/LocalStrings.properties | 1 +
.../catalina/tribes/transport/nio/NioReceiver.java | 1 +
.../tribes/transport/nio/ParallelNioSender.java | 1 +
.../tribes/transport/nio/PooledParallelSender.java | 13 +-
.../tribes/transport/TestPooledSender.java | 135 ++++++++++++++++++++
webapps/docs/changelog.xml | 8 ++
webapps/docs/config/cluster-sender.xml | 5 +-
12 files changed, 242 insertions(+), 78 deletions(-)
diff --git a/java/org/apache/catalina/tribes/group/GroupChannel.java
b/java/org/apache/catalina/tribes/group/GroupChannel.java
index 8e2ff6d2a7..8546d5f805 100644
--- a/java/org/apache/catalina/tribes/group/GroupChannel.java
+++ b/java/org/apache/catalina/tribes/group/GroupChannel.java
@@ -506,7 +506,10 @@ public class GroupChannel extends ChannelInterceptorBase
implements ManagedChann
// There was an error executing the scheduled task, get it and
log it
try {
heartbeatFuture.get();
- } catch (InterruptedException | ExecutionException e) {
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+
log.error(sm.getString("groupChannel.unable.sendHeartbeat"), e);
+ } catch (ExecutionException e) {
log.error(sm.getString("groupChannel.unable.sendHeartbeat"), e);
}
}
diff --git
a/java/org/apache/catalina/tribes/group/interceptors/NonBlockingCoordinator.java
b/java/org/apache/catalina/tribes/group/interceptors/NonBlockingCoordinator.java
index 92169ac670..919dc0ba4f 100644
---
a/java/org/apache/catalina/tribes/group/interceptors/NonBlockingCoordinator.java
+++
b/java/org/apache/catalina/tribes/group/interceptors/NonBlockingCoordinator.java
@@ -51,10 +51,9 @@ import org.apache.juli.logging.LogFactory;
* groups automatically when members are discovered that weren't part of the
* </p>
* <p>
- * This algorithm is non blocking meaning it allows for transactions while the
coordination phase is going on. Note
- * that {@link #startElection(boolean)} itself blocks its caller for up to
- * {@link #getWaitForCoordMsgTimeout()} milliseconds while waiting for an
election request from a higher priority
- * member.
+ * This algorithm is non blocking meaning it allows for transactions while the
coordination phase is going on. Note that
+ * {@link #startElection(boolean)} itself blocks its caller for up to {@link
#getWaitForCoordMsgTimeout()} milliseconds
+ * while waiting for an election request from a higher priority member.
* </p>
* <p>
* This implementation is based on a home brewed algorithm that uses the
AbsoluteOrder of a membership to pass a token
@@ -152,20 +151,17 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
50, 111, 103, 65, 3, -77, 51, -35, 0, 119, 117, 9, -26, 119, 50,
-75, -105, -102, 36, 79, 37, -68, -84,
-123, 15, -22, -109, 106, -55 };
/**
- * Time to wait for coordination timeout. A node that is not the leader in
its view
- * waits up to this duration for a coordination message when an election
is started.
- * The wait blocks the calling thread, which for membership events is the
channel
- * utility executor thread that also runs the channel heartbeat. Lowering
this value
- * reduces the time that a single membership event can delay heartbeats
and other
- * membership processing. The liveness checks performed while merging an
arriving
- * coordination message use {@link #mergeAliveTimeout} instead.
+ * Time to wait for coordination timeout. A node that is not the leader in
its view waits up to this duration for a
+ * coordination message when an election is started. The wait blocks the
calling thread, which for membership events
+ * is the channel utility executor thread that also runs the channel
heartbeat. Lowering this value reduces the time
+ * that a single membership event can delay heartbeats and other
membership processing. The liveness checks
+ * performed while merging an arriving coordination message use {@link
#mergeAliveTimeout} instead.
*/
protected long waitForCoordMsgTimeout = 15000;
/**
- * Time to wait for the TCP connect of the member liveness checks
performed while
- * merging an arriving coordination message. These checks run on the
channel receiver
- * worker threads, so this timeout is kept separate from and much shorter
than
- * {@link #waitForCoordMsgTimeout}.
+ * Time to wait for the TCP connect of the member liveness checks
performed while merging an arriving coordination
+ * message. These checks run on the channel receiver worker threads, so
this timeout is kept separate from and much
+ * shorter than {@link #waitForCoordMsgTimeout}.
*/
protected long mergeAliveTimeout = 1000;
/**
@@ -218,8 +214,8 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
}
/**
- * Returns the time in milliseconds that a non-leader node waits for a
coordination
- * message when an election is started.
+ * Returns the time in milliseconds that a non-leader node waits for a
coordination message when an election is
+ * started.
*
* @return the coordination message wait timeout in milliseconds
*/
@@ -228,8 +224,8 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
}
/**
- * Returns the time in milliseconds that the TCP connect of the member
liveness
- * checks waits during the merge of an arriving coordination message.
+ * Returns the time in milliseconds that the TCP connect of the member
liveness checks waits during the merge of an
+ * arriving coordination message.
*
* @return the merge liveness check timeout in milliseconds
*/
@@ -238,14 +234,12 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
}
/**
- * Sets the time in milliseconds that the TCP connect of the member
liveness checks
- * waits during the merge of an arriving coordination message. Since these
checks run
- * sequentially on the channel receiver worker threads, the value should
stay short to
- * avoid pinning receiver workers when a coordination message carries
stale or dead
+ * Sets the time in milliseconds that the TCP connect of the member
liveness checks waits during the merge of an
+ * arriving coordination message. Since these checks run sequentially on
the channel receiver worker threads, the
+ * value should stay short to avoid pinning receiver workers when a
coordination message carries stale or dead
* members.
*
- * @param mergeAliveTimeout the merge liveness check timeout in
milliseconds, at
- * least one and no more than
+ * @param mergeAliveTimeout the merge liveness check timeout in
milliseconds, at least one and no more than
* {@link Integer#MAX_VALUE}
*/
public void setMergeAliveTimeout(long mergeAliveTimeout) {
@@ -260,22 +254,18 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
}
/**
- * Sets the time in milliseconds that a non-leader node waits for a
coordination
- * message when an election is started. Since the wait blocks the calling
thread,
- * which for membership events is the single threaded channel utility
executor that
- * also runs the channel heartbeat, a lower value limits the delay that a
single
- * membership event can impose on other channel maintenance work. The
merge-time
- * liveness checks are not affected by this value, they use
- * {@link #getMergeAliveTimeout()}.
+ * Sets the time in milliseconds that a non-leader node waits for a
coordination message when an election is
+ * started. Since the wait blocks the calling thread, which for membership
events is the single threaded channel
+ * utility executor that also runs the channel heartbeat, a lower value
limits the delay that a single membership
+ * event can impose on other channel maintenance work. The merge-time
liveness checks are not affected by this
+ * value, they use {@link #getMergeAliveTimeout()}.
*
- * @param waitForCoordMsgTimeout the coordination message wait timeout in
- * milliseconds, at least one and no
more than
- * {@link Integer#MAX_VALUE}
+ * @param waitForCoordMsgTimeout the coordination message wait timeout in
milliseconds, at least one and no more
+ * than {@link Integer#MAX_VALUE}
*/
public void setWaitForCoordMsgTimeout(long waitForCoordMsgTimeout) {
if (waitForCoordMsgTimeout < 1) {
- throw new IllegalArgumentException(
-
sm.getString("nonBlockingCoordinator.waitForCoordMsgTimeout.tooSmall"));
+ throw new
IllegalArgumentException(sm.getString("nonBlockingCoordinator.waitForCoordMsgTimeout.tooSmall"));
}
if (waitForCoordMsgTimeout > Integer.MAX_VALUE) {
throw new
IllegalArgumentException(sm.getString("nonBlockingCoordinator.waitForCoordMsgTimeout.tooLarge",
@@ -292,6 +282,7 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
* Starts an election to determine the coordinator.
*
* @param force Whether to force a new election even if one is in progress
+ *
* @throws ChannelException if an error occurs during the election
*/
public void startElection(boolean force) throws ChannelException {
@@ -353,10 +344,11 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
* happened (i.e. notify() was not called) AND b) the
timeout has not expired AND c) the thread was
* not interrupted.
*/
- } while (suggestedviewId == null && !coordMsgReceived.get() &&
timeout > 0 && !Thread.interrupted());
+ } while (suggestedviewId == null && !coordMsgReceived.get() &&
timeout > 0 &&
+ !Thread.currentThread().isInterrupted());
String msg;
if (suggestedviewId == null && !coordMsgReceived.get()) {
- if (Thread.interrupted()) {
+ if (Thread.currentThread().isInterrupted()) {
msg = "Election abandoned, waiting interrupted.";
} else {
msg = "Election abandoned, waiting timed out.";
@@ -382,8 +374,9 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
* Sends an election message to the next member in the ring.
*
* @param local The local member
- * @param next The next member to send the message to
- * @param msg The coordination message to send
+ * @param next The next member to send the message to
+ * @param msg The coordination message to send
+ *
* @throws ChannelException if sending fails
*/
protected void sendElectionMsg(Member local, Member next,
CoordinationMessage msg) throws ChannelException {
@@ -396,7 +389,8 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
* Sends an election message to the next member inline, retrying on
failure.
*
* @param local The local member
- * @param msg The coordination message to send
+ * @param msg The coordination message to send
+ *
* @throws ChannelException if sending fails to all members
*/
protected void sendElectionMsgToNextInline(Member local,
CoordinationMessage msg) throws ChannelException {
@@ -422,8 +416,9 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
/**
* Creates a ChannelData object from a coordination message.
*
- * @param msg The coordination message
+ * @param msg The coordination message
* @param local The local member
+ *
* @return The channel data containing the message
*/
public ChannelData createData(CoordinationMessage msg, Member local) {
@@ -437,10 +432,11 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
}
/**
- * Checks if a member is alive using the merge liveness timeout
- * ({@link #getMergeAliveTimeout()}) as the connection timeout.
+ * Checks if a member is alive using the merge liveness timeout ({@link
#getMergeAliveTimeout()}) as the connection
+ * timeout.
*
* @param mbr The member to check
+ *
* @return true if the member is alive
*/
protected boolean alive(Member mbr) {
@@ -450,8 +446,9 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
/**
* Checks if a member is alive by attempting a socket connection.
*
- * @param mbr The member to check
+ * @param mbr The member to check
* @param conTimeout The connection timeout in milliseconds
+ *
* @return true if the member is alive
*/
protected boolean memberAlive(Member mbr, long conTimeout) {
@@ -477,6 +474,7 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
* Merges the incoming coordination message members with the current
membership.
*
* @param msg The coordination message
+ *
* @return The merged membership
*/
protected Membership mergeOnArrive(CoordinationMessage msg) {
@@ -504,6 +502,7 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
* Processes an incoming coordination message.
*
* @param msg The coordination message to process
+ *
* @throws ChannelException if processing fails
*/
protected void processCoordMessage(CoordinationMessage msg) throws
ChannelException {
@@ -524,8 +523,9 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
/**
* Handles a coordination token, routing to the appropriate handler based
on source.
*
- * @param msg The coordination message
+ * @param msg The coordination message
* @param merged The merged membership
+ *
* @throws ChannelException if handling fails
*/
protected void handleToken(CoordinationMessage msg, Membership merged)
throws ChannelException {
@@ -541,9 +541,10 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
/**
* Handles a coordination token that originated from the local member.
*
- * @param local The local member
- * @param msg The coordination message
+ * @param local The local member
+ * @param msg The coordination message
* @param merged The merged membership
+ *
* @throws ChannelException if handling fails
*/
protected void handleMyToken(Member local, CoordinationMessage msg,
Membership merged) throws ChannelException {
@@ -573,9 +574,10 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
/**
* Handles a coordination token from another member.
*
- * @param local The local member
- * @param msg The coordination message
+ * @param local The local member
+ * @param msg The coordination message
* @param merged The merged membership
+ *
* @throws ChannelException if handling fails
*/
protected void handleOtherToken(Member local, CoordinationMessage msg,
Membership merged) throws ChannelException {
@@ -591,8 +593,9 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
/**
* Handles a view confirmation message.
*
- * @param msg The coordination message
+ * @param msg The coordination message
* @param merged The merged membership
+ *
* @throws ChannelException if handling fails
*/
protected void handleViewConf(CoordinationMessage msg, Membership merged)
throws ChannelException {
@@ -625,6 +628,7 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
* Checks if a coordination message is a view confirmation.
*
* @param msg The coordination message
+ *
* @return true if the message is a view confirmation
*/
protected boolean isViewConf(CoordinationMessage msg) {
@@ -635,7 +639,8 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
* Checks if the complete membership has a higher priority leader than the
local membership.
*
* @param complete The complete membership
- * @param local The local membership
+ * @param local The local membership
+ *
* @return true if the complete membership has higher priority
*/
protected boolean hasHigherPriority(Member[] complete, Member[] local) {
@@ -789,7 +794,7 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
* Handles the addition of a new member to the group.
*
* @param member The member that was added
- * @param elect Whether to trigger an election
+ * @param elect Whether to trigger an election
*/
public void memberAdded(Member member, boolean elect) {
if (membership == null) {
@@ -917,11 +922,11 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
* Represents a coordination message used in the election protocol.
*/
public static class CoordinationMessage {
- // Message format: X{A-ldr, A-src, mbrs-A,B,C,D}
- // X{A-ldr, A-src, mbrs-A,B,C,D}
- /**
- * The byte buffer for the message.
- */
+ // Message format: X{A-ldr, A-src, mbrs-A,B,C,D}
+ // X{A-ldr, A-src, mbrs-A,B,C,D}
+ /**
+ * The byte buffer for the message.
+ */
protected final XByteBuffer buf;
/**
* The leader member.
@@ -959,9 +964,9 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
*
* @param leader The leader member
* @param source The source member
- * @param view The member view
- * @param id The message ID
- * @param type The message type
+ * @param view The member view
+ * @param id The message ID
+ * @param type The message type
*/
public CoordinationMessage(Member leader, Member source, Member[]
view, UniqueId id, byte[] type) {
this.buf = new XByteBuffer(4096, false);
@@ -1138,8 +1143,8 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
* Represents a coordination event in the election protocol.
*/
public static class CoordinationEvent implements InterceptorEvent {
- // Event type constants for coordination events.
- /** Event for coordinator start. */
+ // Event type constants for coordination events.
+ /** Event for coordinator start. */
public static final int EVT_START = 1;
/** Event for member add. */
public static final int EVT_MBR_ADD = 2;
@@ -1167,7 +1172,7 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
public static final int EVT_ELECT_ABANDONED = 13;
// Event fields: type, interceptor, coord, mbrs, info, view, and
suggestedView.
- /** Event type. */
+ /** Event type. */
final int type;
/** Channel interceptor. */
final ChannelInterceptor interceptor;
@@ -1185,9 +1190,9 @@ public class NonBlockingCoordinator extends
ChannelInterceptorBase {
/**
* Constructs a CoordinationEvent.
*
- * @param type The event type
+ * @param type The event type
* @param interceptor The channel interceptor
- * @param info Additional information about the event
+ * @param info Additional information about the event
*/
public CoordinationEvent(int type, ChannelInterceptor interceptor,
String info) {
this.type = type;
diff --git a/java/org/apache/catalina/tribes/membership/McastServiceImpl.java
b/java/org/apache/catalina/tribes/membership/McastServiceImpl.java
index fda4d55153..ad35ac3268 100644
--- a/java/org/apache/catalina/tribes/membership/McastServiceImpl.java
+++ b/java/org/apache/catalina/tribes/membership/McastServiceImpl.java
@@ -311,7 +311,8 @@ public class McastServiceImpl extends
MembershipProviderBase {
try {
Thread.sleep(memberwait);
} catch (InterruptedException ignore) {
- // Ignore
+ // Stop waiting but leave the interrupt for the caller to see
+ Thread.currentThread().interrupt();
}
if (log.isInfoEnabled()) {
log.info(sm.getString("mcastServiceImpl.waitForMembers.done",
Integer.toString(level)));
@@ -857,6 +858,9 @@ public class McastServiceImpl extends
MembershipProviderBase {
sleep(parent.recoverySleepTime);
}
} catch (InterruptedException ignore) {
+ // The recovery thread has been interrupted so stop
trying to recover
+ Thread.currentThread().interrupt();
+ return;
}
}
} finally {
diff --git
a/java/org/apache/catalina/tribes/membership/StaticMembershipProvider.java
b/java/org/apache/catalina/tribes/membership/StaticMembershipProvider.java
index 0701e2064a..5d158d6723 100644
--- a/java/org/apache/catalina/tribes/membership/StaticMembershipProvider.java
+++ b/java/org/apache/catalina/tribes/membership/StaticMembershipProvider.java
@@ -570,7 +570,7 @@ public class StaticMembershipProvider extends
MembershipProviderBase
sleep(pingInterval);
ping();
} catch (InterruptedException ix) {
- // Ignore
+ // Expected during shutdown. The running flag is already
set to false, so the loop will exit.
} catch (Exception e) {
log.warn(sm.getString("staticMembershipProvider.pingThread.failed"), e);
}
diff --git a/java/org/apache/catalina/tribes/transport/PooledSender.java
b/java/org/apache/catalina/tribes/transport/PooledSender.java
index c4b46a451a..ba846f5f9c 100644
--- a/java/org/apache/catalina/tribes/transport/PooledSender.java
+++ b/java/org/apache/catalina/tribes/transport/PooledSender.java
@@ -210,7 +210,7 @@ public abstract class PooledSender extends AbstractSender
implements MultiPointS
}
public synchronized DataSender getSender(long timeout) {
- long start = System.currentTimeMillis();
+ long start = System.nanoTime();
while (true) {
if (!isOpen) {
throw new
IllegalStateException(sm.getString("pooledSender.closed.queue"));
@@ -227,7 +227,7 @@ public abstract class PooledSender extends AbstractSender
implements MultiPointS
}
try {
if (timeout > 0) {
- long delta = System.currentTimeMillis() - start;
+ long delta = (System.nanoTime() - start) / 1_000_000;
if (delta >= timeout) {
return null;
}
diff --git
a/java/org/apache/catalina/tribes/transport/nio/LocalStrings.properties
b/java/org/apache/catalina/tribes/transport/nio/LocalStrings.properties
index da67c2f2a2..1d98cbda8e 100644
--- a/java/org/apache/catalina/tribes/transport/nio/LocalStrings.properties
+++ b/java/org/apache/catalina/tribes/transport/nio/LocalStrings.properties
@@ -71,4 +71,5 @@ pooledParallelSender.disconnectOnError=Disconnect following
error
pooledParallelSender.sender.disconnected=Sender not connected.
pooledParallelSender.unable.open=Unable to open NIO selector.
pooledParallelSender.unable.retrieveSender=Unable to retrieve a sender from
the sender pool
+pooledParallelSender.unable.retrieveSender.interrupted=Unable to retrieve a
data sender, the thread was interrupted while waiting.
pooledParallelSender.unable.retrieveSender.timeout=Unable to retrieve a data
sender, time out([{0}] ms) error.
diff --git a/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
b/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
index fc5090334e..40863a5833 100644
--- a/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
+++ b/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
@@ -242,6 +242,7 @@ public class NioReceiver extends ReceiverBase implements
Runnable, NioReceiverMB
}
} catch (InterruptedException e) {
log.warn(sm.getString("nioReceiver.stop.executor.interrupted"));
+ Thread.currentThread().interrupt();
}
}
}
diff --git
a/java/org/apache/catalina/tribes/transport/nio/ParallelNioSender.java
b/java/org/apache/catalina/tribes/transport/nio/ParallelNioSender.java
index 719798bf2e..c212c8b0de 100644
--- a/java/org/apache/catalina/tribes/transport/nio/ParallelNioSender.java
+++ b/java/org/apache/catalina/tribes/transport/nio/ParallelNioSender.java
@@ -533,6 +533,7 @@ public class ParallelNioSender extends AbstractSender
implements MultiPointSende
}
} catch (InterruptedException e) {
log.warn(sm.getString("parallelNioSender.disconnect.executor.interrupted"));
+ Thread.currentThread().interrupt();
} finally {
sendSecureExecutor = null;
}
diff --git
a/java/org/apache/catalina/tribes/transport/nio/PooledParallelSender.java
b/java/org/apache/catalina/tribes/transport/nio/PooledParallelSender.java
index 45a69a4b9b..c3b0cca3ca 100644
--- a/java/org/apache/catalina/tribes/transport/nio/PooledParallelSender.java
+++ b/java/org/apache/catalina/tribes/transport/nio/PooledParallelSender.java
@@ -49,11 +49,16 @@ public class PooledParallelSender extends PooledSender
implements PooledParallel
}
ParallelNioSender sender = (ParallelNioSender) getSender();
if (sender == null) {
- ChannelException cx = new ChannelException(
+ // A null sender means either that the wait timed out or that the
thread was interrupted while waiting
+ boolean interrupted = Thread.currentThread().isInterrupted();
+ ChannelException cx = new ChannelException(interrupted ?
+
sm.getString("pooledParallelSender.unable.retrieveSender.interrupted") :
sm.getString("pooledParallelSender.unable.retrieveSender.timeout",
Long.toString(getMaxWait())));
- for (Member member : destination) {
- cx.addFaultyMember(member,
- new
NullPointerException(sm.getString("pooledParallelSender.unable.retrieveSender")));
+ if (!interrupted) {
+ for (Member member : destination) {
+ cx.addFaultyMember(member,
+ new
NullPointerException(sm.getString("pooledParallelSender.unable.retrieveSender")));
+ }
}
throw cx;
} else {
diff --git a/test/org/apache/catalina/tribes/transport/TestPooledSender.java
b/test/org/apache/catalina/tribes/transport/TestPooledSender.java
new file mode 100644
index 0000000000..c6717537bd
--- /dev/null
+++ b/test/org/apache/catalina/tribes/transport/TestPooledSender.java
@@ -0,0 +1,135 @@
+/*
+ * 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.transport;
+
+import java.lang.reflect.Proxy;
+import java.util.concurrent.atomic.AtomicReference;
+
+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.Member;
+
+public class TestPooledSender {
+
+ private static final long TIMEOUT = 10000;
+
+
+ private static class Pool extends PooledSender {
+ @Override
+ public DataSender getNewDataSender() {
+ return (DataSender)
Proxy.newProxyInstance(DataSender.class.getClassLoader(),
+ new Class<?>[] { DataSender.class }, (proxy, method, args)
-> {
+ if (method.getReturnType() == boolean.class) {
+ return Boolean.FALSE;
+ }
+ return null;
+ });
+ }
+
+ @Override
+ public void sendMessage(Member[] destination, ChannelMessage msg)
throws ChannelException {
+ // NO-OP
+ }
+ }
+
+
+ private static Pool createPool(int size, long maxWait) throws Exception {
+ Pool pool = new Pool();
+ pool.setPoolSize(size);
+ pool.setMaxWait(maxWait);
+ pool.connect();
+ return pool;
+ }
+
+
+ private static Thread getSenderInThread(Pool pool,
AtomicReference<DataSender> result,
+ AtomicReference<Boolean> interrupted) {
+ Thread thread = new Thread(() -> {
+ result.set(pool.getSender());
+
interrupted.set(Boolean.valueOf(Thread.currentThread().isInterrupted()));
+ });
+ thread.start();
+ return thread;
+ }
+
+
+ @Test(timeout = 60000)
+ public void testWaitEndsWhenSenderReturned() throws Exception {
+ Pool pool = createPool(1, 0);
+ DataSender first = pool.getSender();
+ Assert.assertNotNull(first);
+
+ AtomicReference<DataSender> result = new AtomicReference<>();
+ Thread waiter = getSenderInThread(pool, result, new
AtomicReference<>());
+ Thread.sleep(500);
+ Assert.assertTrue(waiter.isAlive());
+
+ pool.returnSender(first);
+ waiter.join(TIMEOUT);
+ Assert.assertFalse(waiter.isAlive());
+ Assert.assertSame(first, result.get());
+ }
+
+
+ @Test(timeout = 60000)
+ public void testWaitTimesOut() throws Exception {
+ Pool pool = createPool(1, 500);
+ Assert.assertNotNull(pool.getSender());
+
+ long start = System.nanoTime();
+ Assert.assertNull(pool.getSender());
+ long elapsed = (System.nanoTime() - start) / 1_000_000;
+ Assert.assertTrue("Returned after " + elapsed + "ms", elapsed >= 400);
+ }
+
+
+ @Test(timeout = 60000)
+ public void testInterruptEndsWait() throws Exception {
+ Pool pool = createPool(1, 0);
+ Assert.assertNotNull(pool.getSender());
+
+ AtomicReference<DataSender> result = new AtomicReference<>();
+ AtomicReference<Boolean> interrupted = new AtomicReference<>();
+ Thread waiter = getSenderInThread(pool, result, interrupted);
+ Thread.sleep(500);
+ waiter.interrupt();
+ waiter.join(TIMEOUT);
+ Assert.assertFalse(waiter.isAlive());
+ Assert.assertNull(result.get());
+ Assert.assertEquals(Boolean.TRUE, interrupted.get());
+ }
+
+
+ @Test(timeout = 60000)
+ public void testRaisingLimitEndsWait() throws Exception {
+ Pool pool = createPool(1, 0);
+ Assert.assertNotNull(pool.getSender());
+
+ AtomicReference<DataSender> result = new AtomicReference<>();
+ Thread waiter = getSenderInThread(pool, result, new
AtomicReference<>());
+ Thread.sleep(500);
+ Assert.assertTrue(waiter.isAlive());
+
+ pool.setPoolSize(2);
+ waiter.join(TIMEOUT);
+ Assert.assertFalse(waiter.isAlive());
+ Assert.assertNotNull(result.get());
+ }
+}
diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml
index f956c3e06e..7188f88447 100644
--- a/webapps/docs/changelog.xml
+++ b/webapps/docs/changelog.xml
@@ -520,6 +520,14 @@
so that the configured values are not lost when the configuration is
stored. (markt)
</fix>
+ <fix>
+ Preserve the interrupt status of threads that are interrupted while
+ waiting in the cluster code (sender and receiver shutdown, membership
+ start-up and recovery, heartbeat checks and coordinator elections),
+ report an interrupt while waiting for a pooled sender as such rather
+ than as a time out and measure the wait for a pooled sender with a
+ monotonic clock. (markt)
+ </fix>
</changelog>
</subsection>
<subsection name="WebSocket">
diff --git a/webapps/docs/config/cluster-sender.xml
b/webapps/docs/config/cluster-sender.xml
index 2d3b0651ca..603447a8bb 100644
--- a/webapps/docs/config/cluster-sender.xml
+++ b/webapps/docs/config/cluster-sender.xml
@@ -170,8 +170,9 @@
</attribute>
<attribute name="maxWait" required="false">
The maximum number of milliseconds that the senderPool will wait when
- there are no available senders. The default value is <code>3000</code>
- milliseconds.(3 seconds).
+ there are no available senders. A value of zero or less means that the
+ senderPool will wait until a sender becomes available. The default
+ value is <code>3000</code> milliseconds.(3 seconds).
</attribute>
</attributes>
</subsection>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
