You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@tomcat.apache.org by re...@apache.org on 2026/10/11 13:12:04 UTC
(tomcat) branch 11.0.x updated (11e6d684b7 -> 1b796e1dff)
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a change to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
from 11e6d684b7 Follow-up to "Track StartTlsResponse per connection in JNDIRealm"
new 8393a3cbf1 Validate package size field in XByteBuffer package framing
new c9ede40486 Make the buffer pool size accounting atomic
new f74d9985e6 Avoid NullPointerException when stopping channel components without a registry
new fdca82b2aa Add the null guards of getRegistry to removeRegistry
new 03c16b4c50 Guard the sender recovery trigger against a concurrent stop
new 6adaf3d700 Protect the multicast shutdown send so that stop completes
new f3d134171f Report the configured mcast so timeout and TTL through JMX
new c49f50af35 Bound the multicast send paths to a sendable packet size
new 815d559223 Reject non-positive membership timing properties at start
new 1e8fea97f9 Fail clearly when static membership starts without a local member
new 4f06521931 Warn when static membership properties are replaced while running
new 9507891f30 Keep the derived membership name out of the exposed properties
new ab90082aaa Hand out a copy of the static members list
new 047dd7e9a4 Fail the start when the Kubernetes service account token cannot be read
new 0de06540fa Clamp the backup index to the member snapshot in publishEntryInfo
new 1d5cd7a559 Scope the map member and state locks of AbstractReplicatedMap to the map operations
new 3ff5ca9b1f Send broadcasts and pings to the checked member snapshot
new 523945e31d Skip the removal notification for keys that cannot be serialized
new 56500fdbdf Count entries with a null value in the size of a replicated map
new 1b796e1dff Deserialize the key of a backup request with the loaders of the map
The 20 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../tribes/group/ChannelInterceptorBase.java | 5 +-
.../apache/catalina/tribes/group/GroupChannel.java | 5 +-
java/org/apache/catalina/tribes/io/BufferPool.java | 27 +-
.../org/apache/catalina/tribes/io/XByteBuffer.java | 14 +-
.../apache/catalina/tribes/jmx/JmxRegistry.java | 6 +-
.../tribes/membership/LocalStrings.properties | 4 +
.../catalina/tribes/membership/McastService.java | 39 ++-
.../tribes/membership/McastServiceImpl.java | 24 +-
.../catalina/tribes/membership/MemberImpl.java | 2 +-
.../membership/StaticMembershipProvider.java | 12 +-
.../tribes/membership/StaticMembershipService.java | 36 ++-
.../membership/cloud/CloudMembershipService.java | 5 +-
.../cloud/KubernetesMembershipProvider.java | 7 +-
.../membership/cloud/LocalStrings.properties | 1 +
.../tribes/tipis/AbstractReplicatedMap.java | 118 +++++--
.../catalina/tribes/tipis/LazyReplicatedMap.java | 8 +-
.../catalina/tribes/transport/ReceiverBase.java | 5 +-
.../tribes/transport/ReplicationTransmitter.java | 5 +-
.../catalina/storeconfig/TestStoreConfig.java | 9 +
.../apache/catalina/tribes/io/TestBufferPool.java | 153 ++++++++++
.../apache/catalina/tribes/io/TestXByteBuffer.java | 63 ++++
.../catalina/tribes/jmx/TestJmxRegistry.java | 117 +++++++
.../tribes/membership/TestMcastService.java | 216 +++++++++++++
.../catalina/tribes/membership/TestMemberImpl.java | 48 +++
.../membership/TestStaticMembershipService.java | 145 +++++++++
.../cloud/TestKubernetesMembershipProvider.java | 64 ++++
.../tribes/tipis/TestAbstractReplicatedMap.java | 339 +++++++++++++++++++++
.../tribes/tipis/TestLazyReplicatedMap.java | 61 ++++
webapps/docs/changelog.xml | 12 +
29 files changed, 1474 insertions(+), 76 deletions(-)
create mode 100644 test/org/apache/catalina/tribes/io/TestBufferPool.java
create mode 100644 test/org/apache/catalina/tribes/jmx/TestJmxRegistry.java
create mode 100644 test/org/apache/catalina/tribes/membership/TestMcastService.java
create mode 100644 test/org/apache/catalina/tribes/membership/TestMemberImpl.java
create mode 100644 test/org/apache/catalina/tribes/membership/TestStaticMembershipService.java
create mode 100644 test/org/apache/catalina/tribes/membership/cloud/TestKubernetesMembershipProvider.java
create mode 100644 test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java
create mode 100644 test/org/apache/catalina/tribes/tipis/TestLazyReplicatedMap.java
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
(tomcat) 19/20: Count entries with a null value in the size of a replicated map
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 56500fdbdfdb75070ad041d07163c431463bd107
Author: opencode <re...@apache.org>
AuthorDate: Sun Oct 11 14:45:37 2026 +0200
Count entries with a null value in the size of a replicated map
The size and the values collection only counted active entries that
carried a non-null value, while the entry and key views included every
active entry. A put of a null value is stored as an active local entry,
because the replication gate does not accept a null value, so the size
no longer matched the size of the entry set, which the map contract
requires, and the empty check inherited the inconsistency.
Both views now use the same filter as the entry and key views and count
active entries with a null value, like the standard map implementations
that permit null values do.
---
.../tribes/tipis/AbstractReplicatedMap.java | 4 ++--
.../tribes/tipis/TestAbstractReplicatedMap.java | 22 ++++++++++++++++++++++
2 files changed, 24 insertions(+), 2 deletions(-)
diff --git a/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java b/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
index ead1263dd8..ac43555133 100644
--- a/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
+++ b/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
@@ -1537,7 +1537,7 @@ public abstract class AbstractReplicatedMap<K, V>
int counter = 0;
for (Entry<K,MapEntry<K,V>> e : innerMap.entrySet()) {
MapEntry<K,V> entry = e.getValue();
- if (entry != null && entry.isActive() && entry.getValue() != null) {
+ if (entry != null && entry.isActive()) {
counter++;
}
}
@@ -1564,7 +1564,7 @@ public abstract class AbstractReplicatedMap<K, V>
List<V> values = new ArrayList<>();
for (Entry<K,MapEntry<K,V>> e : innerMap.entrySet()) {
MapEntry<K,V> entry = e.getValue();
- if (entry != null && entry.isActive() && entry.getValue() != null) {
+ if (entry != null && entry.isActive()) {
values.add(entry.getValue());
}
}
diff --git a/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java b/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java
index e4da5f26fa..ae82476a55 100644
--- a/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java
+++ b/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java
@@ -244,4 +244,26 @@ public class TestAbstractReplicatedMap {
Assert.assertNull(map.getInternal(key));
Assert.assertEquals(0, channel.sendCalls.get());
}
+
+ /*
+ * A put stores an entry with a null value, so the size must count it like the entry and key
+ * views do. The size previously filtered such an entry out, breaking the map contract that the
+ * size equals the size of the entry set, and the isEmpty inherited the inconsistency.
+ */
+ @Test
+ public void testSizeCountsActiveEntriesWithNullValue() throws Exception {
+ LazyReplicatedMap<String,String> map = new LazyReplicatedMap<String,String>(
+ (key, value) -> {
+ // Not made primary here
+ }, new TestChannel(), 1000, "TestAbstractReplicatedMap", null);
+ map.put("key", "value");
+ map.put("other", null);
+
+ Assert.assertEquals(2, map.size());
+ Assert.assertEquals(2, map.entrySet().size());
+ Assert.assertEquals(2, map.keySet().size());
+ Assert.assertEquals(2, map.values().size());
+ Assert.assertFalse(map.isEmpty());
+ Assert.assertTrue(map.values().contains(null));
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
(tomcat) 12/20: Keep the derived membership name out of the exposed properties
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 9507891f307ff1da65526523cdf745e0aec9f84d
Author: opencode <re...@apache.org>
AuthorDate: Sun Oct 11 12:24:05 2026 +0200
Keep the derived membership name out of the exposed properties
The membership provider is passed the derived membership name through
the properties, but the builder injected the internal key into the
very object that the service exposes through its properties, mixing a
derived value into the administrator view of the configuration and
into anything built from a dump or a reapply of it.
The key now goes to the provider on a clone of the configuration,
which also keeps the lookups through caller supplied defaults intact,
while the exposed properties stay exactly as they were configured.
Add a test that the exposed properties stay free of the internal key
and that the start itself proves the provider received the name, for
building the provider without it fails.
---
.../tribes/membership/StaticMembershipService.java | 7 ++++--
.../membership/TestStaticMembershipService.java | 25 ++++++++++++++++++++++
2 files changed, 30 insertions(+), 2 deletions(-)
diff --git a/java/org/apache/catalina/tribes/membership/StaticMembershipService.java b/java/org/apache/catalina/tribes/membership/StaticMembershipService.java
index 85fc6b9ae0..b4d9d3101e 100644
--- a/java/org/apache/catalina/tribes/membership/StaticMembershipService.java
+++ b/java/org/apache/catalina/tribes/membership/StaticMembershipService.java
@@ -97,8 +97,11 @@ public class StaticMembershipService extends MembershipServiceBase implements St
provider.setMembershipListener(this);
provider.setMembershipService(this);
provider.setStaticMembers(staticMembers);
- properties.setProperty("membershipName", getMembershipName());
- provider.init(properties);
+ // Pass the derived membership name to the provider on a copy of the configuration, so the
+ // internal key does not leak into the properties this service exposes
+ Properties providerProperties = (Properties) properties.clone();
+ providerProperties.setProperty("membershipName", getMembershipName());
+ provider.init(providerProperties);
return provider;
}
diff --git a/test/org/apache/catalina/tribes/membership/TestStaticMembershipService.java b/test/org/apache/catalina/tribes/membership/TestStaticMembershipService.java
index a72fc0b7c7..ddafa78835 100644
--- a/test/org/apache/catalina/tribes/membership/TestStaticMembershipService.java
+++ b/test/org/apache/catalina/tribes/membership/TestStaticMembershipService.java
@@ -95,4 +95,29 @@ public class TestStaticMembershipService {
}
Assert.assertNull(service.getMembershipProvider());
}
+
+ /*
+ * The membership name is derived for the provider alone. It must not leak into the
+ * configuration properties that the service exposes, while the provider must still receive
+ * it, which the start itself proves, as it would fail without the name.
+ */
+ @Test
+ public void testMembershipNameStaysOutOfTheExposedProperties() throws Exception {
+ GroupChannel channel = new GroupChannel();
+ channel.setName("TestStaticMembershipService-" + System.nanoTime());
+
+ StaticMembershipService service = new StaticMembershipService();
+ service.setChannel(channel);
+ service.addStaticMember(new StaticMember("127.0.0.1", 4004, -1));
+ service.setLocalMemberProperties("127.0.0.1", 4004, 0, 0);
+
+ int level = Channel.MBR_RX_SEQ | Channel.MBR_TX_SEQ;
+ service.start(level);
+ try {
+ Assert.assertFalse("the internal membership name must not leak into the properties",
+ service.getProperties().containsKey("membershipName"));
+ } finally {
+ service.stop(level);
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
(tomcat) 20/20: Deserialize the key of a backup request with the loaders of the map
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 1b796e1dffc309463b4766f56a4d3980c14d8c60
Author: opencode <re...@apache.org>
AuthorDate: Sun Oct 11 14:53:49 2026 +0200
Deserialize the key of a backup request with the loaders of the map
Answering a backup request read the key of the received message without
specifying class loaders, unlike every other receive path which
deserializes with the loaders of the map. A key whose class is not
visible to the default loaders then failed the reply with an unchecked
exception on the responder, which left the requester without an answer
until its rpc timeout.
The request is now deserialized with the loaders of the map before the
lookup, and a key that cannot be deserialized fails the request with a
plain error log and no reply, the same outcome as for a missing entry.
---
.../tribes/tipis/AbstractReplicatedMap.java | 11 +++-
.../tribes/tipis/TestAbstractReplicatedMap.java | 70 ++++++++++++++++++++++
2 files changed, 80 insertions(+), 1 deletion(-)
diff --git a/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java b/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
index ac43555133..018cadd4d5 100644
--- a/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
+++ b/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
@@ -701,7 +701,16 @@ public abstract class AbstractReplicatedMap<K, V>
// backup request
if (mapmsg.getMsgType() == MapMessage.MSG_RETRIEVE_BACKUP) {
- MapEntry<K,V> entry = innerMap.get(mapmsg.getKey());
+ MapEntry<K,V> entry;
+ try {
+ // Deserialize the key with the external loaders, like the other receive paths do,
+ // since the default loaders cannot see the class of a key from an application
+ mapmsg.deserialize(getExternalLoaders());
+ entry = innerMap.get(mapmsg.getKey());
+ } catch (ClassNotFoundException | IOException x) {
+ log.error(sm.getString("abstractReplicatedMap.unable.deserialize.MapMessage"), x);
+ return null;
+ }
if (entry == null || (!entry.isSerializable())) {
return null;
}
diff --git a/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java b/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java
index ae82476a55..0f3673def6 100644
--- a/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java
+++ b/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java
@@ -35,6 +35,7 @@ import org.apache.catalina.tribes.ErrorHandler;
import org.apache.catalina.tribes.Member;
import org.apache.catalina.tribes.UniqueId;
import org.apache.catalina.tribes.group.GroupChannel;
+import org.apache.catalina.tribes.io.XByteBuffer;
import org.apache.catalina.tribes.membership.MemberImpl;
import org.apache.catalina.tribes.tipis.AbstractReplicatedMap.MapMessage;
@@ -266,4 +267,73 @@ public class TestAbstractReplicatedMap {
Assert.assertFalse(map.isEmpty());
Assert.assertTrue(map.values().contains(null));
}
+
+ private static final class SerialKey implements Serializable {
+
+ @Serial
+ private static final long serialVersionUID = 1L;
+
+ private final String id;
+
+ SerialKey(String id) {
+ this.id = id;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ return o instanceof SerialKey && id.equals(((SerialKey) o).id);
+ }
+
+ @Override
+ public int hashCode() {
+ return id.hashCode();
+ }
+ }
+
+ /*
+ * A received backup request carries its key only in serialized form. The reply must deserialize
+ * that key with the loaders of the map, like the other receive paths do, and must fail the
+ * request without throwing when the key cannot be deserialized, so that the requester gets a
+ * normal timeout instead of being answered by a failure of the responder.
+ */
+ @Test
+ public void testBackupRequestDeserializesKeyOfReceivedMessage() throws Exception {
+ LazyReplicatedMap<Object,String> map = new LazyReplicatedMap<Object,String>(
+ (key, value) -> {
+ // Not made primary here
+ }, new TestChannel(), 1000, "TestAbstractReplicatedMap", null);
+ map.put(new SerialKey("key"), "value");
+ Member sender = new MemberImpl("localhost", 4017, -1);
+
+ // Round trip over the serialization to get a message with the key data but no live key
+ byte[] wire = XByteBuffer.serialize(new MapMessage(
+ "TestAbstractReplicatedMap".getBytes(StandardCharsets.UTF_8), MapMessage.MSG_RETRIEVE_BACKUP, false,
+ new SerialKey("key"), null, null, null, null));
+
+ MapMessage received = (MapMessage) XByteBuffer.deserialize(wire);
+ Serializable reply = map.replyRequest(received, sender);
+ Assert.assertTrue(reply instanceof MapMessage);
+ Assert.assertEquals("value", ((MapMessage) reply).getValue());
+
+ // Corrupt the class name inside the serialized key, so it cannot be deserialized
+ byte[] corrupted = wire.clone();
+ int at = indexOf(corrupted, "SerialKey".getBytes(StandardCharsets.US_ASCII));
+ Assert.assertTrue(at > 0);
+ corrupted[at] = 'X';
+ MapMessage broken = (MapMessage) XByteBuffer.deserialize(corrupted);
+ Assert.assertNull(map.replyRequest(broken, sender));
+ }
+
+ private static int indexOf(byte[] data, byte[] pattern) {
+ outer:
+ for (int i = 0; i <= data.length - pattern.length; i++) {
+ for (int j = 0; j < pattern.length; j++) {
+ if (data[i + j] != pattern[j]) {
+ continue outer;
+ }
+ }
+ return i;
+ }
+ return -1;
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
(tomcat) 14/20: Fail the start when the Kubernetes service account token cannot be read
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 047dd7e9a4b605fa2e6bb0557a47956202e40b9c
Author: opencode <re...@apache.org>
AuthorDate: Sun Oct 11 12:50:34 2026 +0200
Fail the start when the Kubernetes service account token cannot be read
Reading the service account token at start was wrapped in a log of the
failure, after which the service kept running with a provider without
a stream provider. No path ever assigns the stream provider again and
the fetch short circuits before any token reload could run, so such a
node stayed out of the membership for its whole lifetime, while the
other nodes saw its pod running through the API and kept replicating
session data to it. Replication then went one way only, with the loss
of that node sessions on failover as the price, signalled by nothing
but one error log at startup.
The start now fails closed when the token cannot be read, naming the
token path, as the client certificate setup already does for its own
errors, so an unusable node is seen instead of trusted.
Adapt the storeconfig test, which started a cluster with this provider
outside of any pod, to start only the channel receiver and sender. The
restrictions are stored with the cluster, so the restart from the
written configuration keeps the membership service unstarted too, which
the test now asserts. Add a test that an unreadable token fails the
start with the io failure as its cause.
---
.../cloud/KubernetesMembershipProvider.java | 7 ++-
.../membership/cloud/LocalStrings.properties | 1 +
.../catalina/storeconfig/TestStoreConfig.java | 9 +++
.../cloud/TestKubernetesMembershipProvider.java | 64 ++++++++++++++++++++++
4 files changed, 80 insertions(+), 1 deletion(-)
diff --git a/java/org/apache/catalina/tribes/membership/cloud/KubernetesMembershipProvider.java b/java/org/apache/catalina/tribes/membership/cloud/KubernetesMembershipProvider.java
index bdac124c59..b0c76371f5 100644
--- a/java/org/apache/catalina/tribes/membership/cloud/KubernetesMembershipProvider.java
+++ b/java/org/apache/catalina/tribes/membership/cloud/KubernetesMembershipProvider.java
@@ -101,7 +101,12 @@ public class KubernetesMembershipProvider extends CloudMembershipProvider {
streamProvider = new TokenStreamProvider(new String(bytes, StandardCharsets.US_ASCII), caCertFile);
saTokenLastModifiedTime = Files.getLastModifiedTime(saTokenPath);
} catch (IOException ioe) {
- log.error(sm.getString("kubernetesMembershipProvider.streamError"), ioe);
+ // Without the token, the Kubernetes API cannot be accessed and a provider without a
+ // stream provider never recovers, keeping this node out of the membership for its
+ // whole lifetime while the other nodes keep replicating to it. Fail the start
+ // instead, as the client certificate branch does for its own errors.
+ throw new IllegalStateException(sm.getString(
+ "kubernetesMembershipProvider.serviceAccountTokenReadFailed", saTokenFile), ioe);
}
} else {
String clientKeyFile = getEnv("KUBERNETES_CLIENT_KEY_FILE");
diff --git a/java/org/apache/catalina/tribes/membership/cloud/LocalStrings.properties b/java/org/apache/catalina/tribes/membership/cloud/LocalStrings.properties
index 0a0816cf02..1b18dd9f7a 100644
--- a/java/org/apache/catalina/tribes/membership/cloud/LocalStrings.properties
+++ b/java/org/apache/catalina/tribes/membership/cloud/LocalStrings.properties
@@ -41,6 +41,7 @@ kubernetesMembershipProvider.memberError=Error creating member
kubernetesMembershipProvider.noKey=Client certificate key was not specified in the environment
kubernetesMembershipProvider.noNamespace=Namespace not set
kubernetesMembershipProvider.serviceAccountTokenMissing=Service account token not found at [{0}]
+kubernetesMembershipProvider.serviceAccountTokenReadFailed=Unable to read the service account token at [{0}]. The cluster cannot start without access to the Kubernetes API.
kubernetesMembershipProvider.streamError=Failed to open stream
tokenStream.failedConnection=Failed connection to [{0}]
diff --git a/test/org/apache/catalina/storeconfig/TestStoreConfig.java b/test/org/apache/catalina/storeconfig/TestStoreConfig.java
index 5f922cdd2c..0420d4e0e9 100644
--- a/test/org/apache/catalina/storeconfig/TestStoreConfig.java
+++ b/test/org/apache/catalina/storeconfig/TestStoreConfig.java
@@ -107,6 +107,12 @@ public class TestStoreConfig extends TomcatBaseTest {
tc_CloudMembershipService_53.setMembershipProviderClassName("org.apache.catalina.tribes.membership.cloud.KubernetesMembershipProvider");
tc_GroupChannel_52.setMembershipService(tc_CloudMembershipService_53);
tc_SimpleTcpCluster_51.setChannel(tc_GroupChannel_52);
+ // Cloud membership needs the Kubernetes API when it starts. Start the channel's receiver
+ // and sender only, so the storeconfig round-trip does not depend on the API. The start
+ // options are not default, so they are stored in server.xml and apply to the restart from
+ // the written configuration as well.
+ tc_SimpleTcpCluster_51.setChannelStartOptions(
+ org.apache.catalina.tribes.Channel.SND_RX_SEQ | org.apache.catalina.tribes.Channel.SND_TX_SEQ);
tomcat.getEngine().setCluster(tc_SimpleTcpCluster_51);
org.apache.catalina.realm.LockOutRealm tc_LockOutRealm_55 = new org.apache.catalina.realm.LockOutRealm();
@@ -153,6 +159,9 @@ public class TestStoreConfig extends TomcatBaseTest {
}
Assert.assertTrue(serverXmlDump.contains("StoreConfigLifecycleListener"));
Assert.assertTrue(serverXmlDump.contains("UserDatabaseRealm"));
+ // The cluster must round-trip the channel start options the test had to restrict, as the
+ // restart from the written configuration must keep the membership service unstarted
+ Assert.assertTrue(serverXmlDump.contains("channelStartOptions"));
Assert.assertTrue(serverXmlDump.contains("SecretKeyCredentialHandler"));
Assert.assertTrue(serverXmlDump.contains("certificateKeystorePassword="));
Assert.assertTrue(serverXmlDump.contains("+TLSv1.1"));
diff --git a/test/org/apache/catalina/tribes/membership/cloud/TestKubernetesMembershipProvider.java b/test/org/apache/catalina/tribes/membership/cloud/TestKubernetesMembershipProvider.java
new file mode 100644
index 0000000000..b54cae7de6
--- /dev/null
+++ b/test/org/apache/catalina/tribes/membership/cloud/TestKubernetesMembershipProvider.java
@@ -0,0 +1,64 @@
+/*
+ * 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.membership.cloud;
+
+import java.io.IOException;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+
+import org.junit.Assert;
+import org.junit.Assume;
+import org.junit.Test;
+
+import org.apache.catalina.tribes.group.GroupChannel;
+
+public class TestKubernetesMembershipProvider {
+
+ /*
+ * When the service account token cannot be read, the provider has no way to access the
+ * Kubernetes API and never recovers, as the stream provider is only created at start. Such a
+ * node stays out of the membership for its whole lifetime while the other nodes see its pod
+ * and keep replicating to it, turning replication one way. The start must therefore fail
+ * closed, like it does when the client certificate setup fails.
+ */
+ @Test
+ public void testStartFailsWhenServiceAccountTokenUnreadable() throws Exception {
+ // The provider reads the default token file when the environment does not redirect it, so
+ // the test needs an environment without the file and without a redirect to another one
+ Assume.assumeTrue(System.getenv(CloudMembershipProvider.CUSTOM_ENV_PREFIX + "SA_TOKEN_FILE") == null);
+ Assume.assumeTrue(System.getenv("SA_TOKEN_FILE") == null);
+ Assume.assumeTrue(
+ System.getenv(CloudMembershipProvider.CUSTOM_ENV_PREFIX + "CLIENT_CERT_FILE") == null);
+ Assume.assumeTrue(System.getenv("KUBERNETES_CLIENT_CERTIFICATE_FILE") == null);
+ Assume.assumeFalse(Files.exists(
+ FileSystems.getDefault().getPath("/var/run/secrets/kubernetes.io/serviceaccount/token")));
+
+ GroupChannel channel = new GroupChannel();
+ channel.setName("TestKubernetesMembershipProvider-" + System.nanoTime());
+
+ CloudMembershipService service = new CloudMembershipService();
+ service.setChannel(channel);
+ service.setLocalMemberProperties("127.0.0.1", 4006, 0, 0);
+ try {
+ service.start(CloudMembershipService.MBR_RX);
+ Assert.fail("start() must fail when the service account token cannot be read");
+ } catch (IllegalStateException expected) {
+ // The start fails closed rather than running without access to the API
+ Assert.assertTrue(expected.getCause() instanceof IOException);
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
(tomcat) 06/20: Protect the multicast shutdown send so that stop completes
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 6adaf3d700f169aa44c9b6ff1c157e1d84162d2a
Author: opencode <re...@apache.org>
AuthorDate: Sun Oct 11 10:41:52 2026 +0200
Protect the multicast shutdown send so that stop completes
When the last service level was stopped, the SHUTDOWN broadcast was
sent outside the protection that the surrounding leaveGroup and
close calls enjoy. The send routinely fails in exactly the
circumstances that accompany a shutdown, a down network device or an
already broken socket, and its IOException then skipped leaving the
multicast group and closing the socket, and kept the facade from
releasing its implementation, leaking the socket and the group
membership for the rest of the JVM's life. A recovery would retry the
close on the next attempt, but a plain cluster shutdown has no retry.
The send is now wrapped like its neighbours, logging at debug while
the shutdown message simply does not reach the peers, as it already
did not, and they expire the member on the drop time as usual.
Add a test that breaks the socket before stopping and checks that the
implementation is released, which fails on the previous code.
---
.../tribes/membership/McastServiceImpl.java | 9 ++-
.../tribes/membership/TestMcastService.java | 72 ++++++++++++++++++++++
2 files changed, 80 insertions(+), 1 deletion(-)
diff --git a/java/org/apache/catalina/tribes/membership/McastServiceImpl.java b/java/org/apache/catalina/tribes/membership/McastServiceImpl.java
index d5c538026f..24b2b99a19 100644
--- a/java/org/apache/catalina/tribes/membership/McastServiceImpl.java
+++ b/java/org/apache/catalina/tribes/membership/McastServiceImpl.java
@@ -342,7 +342,14 @@ public class McastServiceImpl extends MembershipProviderBase {
if (startLevel == 0) {
// send a stop message
member.setCommand(Member.SHUTDOWN_PAYLOAD);
- send(false);
+ try {
+ send(false);
+ } catch (Exception e) {
+ // Shutting down. Only log at debug.
+ if (log.isDebugEnabled()) {
+ log.debug(sm.getString("mcastServiceImpl.error.stop"), e);
+ }
+ }
// leave mcast group
try {
socket.leaveGroup(new InetSocketAddress(address, 0), null);
diff --git a/test/org/apache/catalina/tribes/membership/TestMcastService.java b/test/org/apache/catalina/tribes/membership/TestMcastService.java
new file mode 100644
index 0000000000..ad5b981db5
--- /dev/null
+++ b/test/org/apache/catalina/tribes/membership/TestMcastService.java
@@ -0,0 +1,72 @@
+/*
+ * 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.membership;
+
+import java.io.ByteArrayInputStream;
+import java.util.Properties;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import org.apache.catalina.tribes.Channel;
+import org.apache.catalina.tribes.group.GroupChannel;
+
+public class TestMcastService {
+
+ /*
+ * The shutdown message that stop() broadcasts routinely fails when the network device is down
+ * or the socket is already broken. That must not skip leaving the multicast group and closing
+ * the socket, nor prevent the service facade from releasing its implementation.
+ */
+ @Test(timeout = 60000)
+ public void testStopWithFailingShutdownSendCompletes() throws Exception {
+ int base = 40000 + (int) Math.floorMod(System.nanoTime(), 10000);
+ StringBuilder properties = new StringBuilder();
+ properties.append("mcastAddress=228.0.0.4\n");
+ properties.append("mcastPort=").append(base).append("\n");
+ properties.append("tcpListenHost=127.0.0.1\n");
+ properties.append("tcpListenPort=").append(base + 1).append("\n");
+ properties.append("tcpSecurePort=0\n");
+ properties.append("udpListenPort=0\n");
+ properties.append("mcastFrequency=100\n");
+ properties.append("memberDropTime=3000\n");
+ properties.append("recoveryEnabled=false\n");
+ properties.append("recoveryCounter=10\n");
+ properties.append("recoverySleepTime=3000\n");
+
+ GroupChannel channel = new GroupChannel();
+ channel.setName("TestMcastService-" + System.nanoTime());
+
+ McastService service = new McastService();
+ service.setChannel(channel);
+ Properties props = new Properties();
+ props.load(new ByteArrayInputStream(properties.toString().getBytes("UTF-8")));
+ service.setProperties(props);
+ service.start(Channel.MBR_TX_SEQ);
+
+ McastServiceImpl impl = service.impl;
+ Assert.assertNotNull(impl);
+
+ // Break the socket so that the shutdown message cannot be sent
+ impl.socket.close();
+
+ // Stop must complete and release the implementation despite the failing shutdown send
+ service.stop(Channel.MBR_TX_SEQ);
+
+ Assert.assertNull("the membership implementation must be released on stop", service.impl);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
(tomcat) 08/20: Bound the multicast send paths to a sendable packet size
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit c49f50af35420ed729367658e6df7cb54720278e
Author: opencode <re...@apache.org>
AuthorDate: Sun Oct 11 11:06:03 2026 +0200
Bound the multicast send paths to a sendable packet size
The size checks that guard the two multicast send paths, the
application broadcasts of the membership service and the member ping
payload, capped the packet at the 65535 byte protocol limit of an IP
packet. That limit includes the IP and UDP headers, so a packet of
65508 bytes or more passed the checks and then failed in the socket
send with a message too long error, driving the membership sender into
its error and recovery loop or surfacing the failure to the
application as a broadcast exception.
The checks now use a maximum send size that leaves room for the
headers and for IP options. The receive buffer is unaffected, as a
receive buffer may be larger than anything the socket will carry.
Add tests for both paths at the new limit, rejecting one byte above it
while the packet is still small enough for the socket to send, so the
checks are shown to reject what the socket could not carry.
---
.../catalina/tribes/membership/McastService.java | 4 +-
.../tribes/membership/McastServiceImpl.java | 7 +++
.../catalina/tribes/membership/MemberImpl.java | 2 +-
.../tribes/membership/TestMcastService.java | 57 ++++++++++++++++++++++
.../catalina/tribes/membership/TestMemberImpl.java | 48 ++++++++++++++++++
5 files changed, 115 insertions(+), 3 deletions(-)
diff --git a/java/org/apache/catalina/tribes/membership/McastService.java b/java/org/apache/catalina/tribes/membership/McastService.java
index 0672086983..0a05ed2f4d 100644
--- a/java/org/apache/catalina/tribes/membership/McastService.java
+++ b/java/org/apache/catalina/tribes/membership/McastService.java
@@ -466,9 +466,9 @@ public class McastService extends MembershipServiceBase implements MessageListen
}
byte[] data = XByteBuffer.createDataPackage((ChannelData) message);
- if (data.length > McastServiceImpl.MAX_PACKET_SIZE) {
+ if (data.length > McastServiceImpl.MAX_SEND_SIZE) {
throw new ChannelException(sm.getString("mcastService.exceed.maxPacketSize", Integer.toString(data.length),
- Integer.toString(McastServiceImpl.MAX_PACKET_SIZE)));
+ Integer.toString(McastServiceImpl.MAX_SEND_SIZE)));
}
DatagramPacket packet = new DatagramPacket(data, 0, data.length);
try {
diff --git a/java/org/apache/catalina/tribes/membership/McastServiceImpl.java b/java/org/apache/catalina/tribes/membership/McastServiceImpl.java
index 24b2b99a19..60935bb95f 100644
--- a/java/org/apache/catalina/tribes/membership/McastServiceImpl.java
+++ b/java/org/apache/catalina/tribes/membership/McastServiceImpl.java
@@ -53,6 +53,13 @@ public class McastServiceImpl extends MembershipProviderBase {
*/
protected static final int MAX_PACKET_SIZE = 65535;
+ /**
+ * The maximum payload size of a packet that can be sent. This is lower than the maximum
+ * packet size because the protocol limit for a UDP datagram includes the IP and UDP headers,
+ * and the IP header may be extended by options.
+ */
+ protected static final int MAX_SEND_SIZE = MAX_PACKET_SIZE - 48;
+
/**
* The string manager for this class.
*/
diff --git a/java/org/apache/catalina/tribes/membership/MemberImpl.java b/java/org/apache/catalina/tribes/membership/MemberImpl.java
index ded0596239..edac5f8dab 100644
--- a/java/org/apache/catalina/tribes/membership/MemberImpl.java
+++ b/java/org/apache/catalina/tribes/membership/MemberImpl.java
@@ -688,7 +688,7 @@ public class MemberImpl implements Member, java.io.Externalizable {
if (newPayloadLength > oldPayloadLength) {
// It is possible that the max packet size will be exceeded
if ((newPayloadLength - oldPayloadLength +
- getData(false, false).length) > McastServiceImpl.MAX_PACKET_SIZE) {
+ getData(false, false).length) > McastServiceImpl.MAX_SEND_SIZE) {
throw new IllegalArgumentException(sm.getString("memberImpl.large.payload"));
}
}
diff --git a/test/org/apache/catalina/tribes/membership/TestMcastService.java b/test/org/apache/catalina/tribes/membership/TestMcastService.java
index 431e6fae9a..77e43af28c 100644
--- a/test/org/apache/catalina/tribes/membership/TestMcastService.java
+++ b/test/org/apache/catalina/tribes/membership/TestMcastService.java
@@ -23,7 +23,10 @@ import org.junit.Assert;
import org.junit.Test;
import org.apache.catalina.tribes.Channel;
+import org.apache.catalina.tribes.ChannelException;
import org.apache.catalina.tribes.group.GroupChannel;
+import org.apache.catalina.tribes.io.ChannelData;
+import org.apache.catalina.tribes.io.XByteBuffer;
public class TestMcastService {
@@ -119,4 +122,58 @@ public class TestMcastService {
Assert.assertEquals(0, service.getSoTimeout());
Assert.assertEquals(0, service.getTtl());
}
+
+ /*
+ * The 65535 byte protocol limit of an IP packet includes the IP and UDP headers, so a
+ * broadcast that fits within it can still exceed what the socket is able to send. Such a
+ * broadcast must be rejected by the size check rather than fail later in the send.
+ */
+ @Test(timeout = 60000)
+ public void testBroadcastRejectsPacketsTheSocketCannotCarry() throws Exception {
+ int base = 40000 + (int) Math.floorMod(System.nanoTime(), 10000);
+ Properties props = new Properties();
+ props.setProperty("mcastAddress", "228.0.0.4");
+ props.setProperty("mcastPort", String.valueOf(base));
+ props.setProperty("tcpListenHost", "127.0.0.1");
+ props.setProperty("tcpListenPort", String.valueOf(base + 1));
+ props.setProperty("tcpSecurePort", "0");
+ props.setProperty("udpListenPort", "0");
+ props.setProperty("mcastFrequency", "100");
+ props.setProperty("memberDropTime", "3000");
+ props.setProperty("recoveryEnabled", "false");
+ props.setProperty("recoveryCounter", "10");
+ props.setProperty("recoverySleepTime", "3000");
+
+ GroupChannel channel = new GroupChannel();
+ channel.setName("TestMcastService-" + System.nanoTime());
+
+ McastService service = new McastService();
+ service.setChannel(channel);
+ service.setProperties(props);
+ service.start(Channel.MBR_TX_SEQ);
+ try {
+ // Build a packet just above the sendable size but still within the 65535 byte packet
+ // limit the old check used, and small enough for this platform to actually send
+ ChannelData probe = new ChannelData();
+ probe.setMessage(new XByteBuffer(new byte[0], false));
+ int overhead = XByteBuffer.createDataPackage(probe).length;
+ Assert.assertTrue(overhead > 0 && overhead < 128);
+ byte[] messageBytes = new byte[McastServiceImpl.MAX_SEND_SIZE + 8 - overhead];
+ Assert.assertEquals(McastServiceImpl.MAX_SEND_SIZE + 8,
+ XByteBuffer.createDataPackage(new ChannelData(probe.getUniqueId(),
+ new XByteBuffer(messageBytes, false), 0)).length);
+
+ ChannelData message = new ChannelData();
+ message.setMessage(new XByteBuffer(messageBytes, false));
+ message.setTimestamp(System.currentTimeMillis());
+ try {
+ service.broadcast(message);
+ Assert.fail("A broadcast above the sendable size must be rejected");
+ } catch (ChannelException expected) {
+ // The size check must reject it
+ }
+ } finally {
+ service.stop(Channel.MBR_TX_SEQ);
+ }
+ }
}
diff --git a/test/org/apache/catalina/tribes/membership/TestMemberImpl.java b/test/org/apache/catalina/tribes/membership/TestMemberImpl.java
new file mode 100644
index 0000000000..b13e8e9030
--- /dev/null
+++ b/test/org/apache/catalina/tribes/membership/TestMemberImpl.java
@@ -0,0 +1,48 @@
+/*
+ * 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.membership;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import org.apache.catalina.tribes.util.UUIDGenerator;
+
+public class TestMemberImpl {
+
+ /*
+ * The 65535 byte protocol limit of an IP packet includes the IP and UDP headers, so a member
+ * ping that fits within it can still be too large for the multicast socket to send. The payload
+ * size check must bound the ping at the size that can actually be sent, not the packet limit.
+ */
+ @Test
+ public void testPayloadIsBoundedAtTheSendableSize() throws Exception {
+ MemberImpl member = new MemberImpl("localhost", 4000, 100);
+ member.setUniqueId(UUIDGenerator.randomUUID(false));
+ int base = member.getData(false, false).length;
+
+ // A payload that fills the ping exactly up to the sendable size is accepted
+ member.setPayload(new byte[McastServiceImpl.MAX_SEND_SIZE - base]);
+
+ // One byte more passes the 65535 byte packet limit but cannot be carried by the socket
+ try {
+ member.setPayload(new byte[McastServiceImpl.MAX_SEND_SIZE - base + 1]);
+ Assert.fail("A payload that takes the ping above the sendable size must be rejected");
+ } catch (IllegalArgumentException expected) {
+ // The size check must reject it
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
(tomcat) 04/20: Add the null guards of getRegistry to removeRegistry
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit fdca82b2aac91677bb467465026fdc1fb9fcc1c8
Author: opencode <re...@apache.org>
AuthorDate: Sun Oct 11 10:25:27 2026 +0200
Add the null guards of getRegistry to removeRegistry
removeRegistry() called getName() on the channel it was handed and
looked the name up in a concurrent hash map, so a null channel or a
channel without a name threw a NullPointerException instead of doing
nothing. getRegistry() explicitly rejects both cases and the public
static API of the registry degrades gracefully to a no-op when JMX is
not configured, so removal should not be stricter than lookup.
Both cases now return early, documented on the method, and a test
covers the null channel as well as the missing name.
---
java/org/apache/catalina/tribes/jmx/JmxRegistry.java | 6 +++++-
test/org/apache/catalina/tribes/jmx/TestJmxRegistry.java | 14 ++++++++++++++
2 files changed, 19 insertions(+), 1 deletion(-)
diff --git a/java/org/apache/catalina/tribes/jmx/JmxRegistry.java b/java/org/apache/catalina/tribes/jmx/JmxRegistry.java
index 1803072961..37ac357824 100644
--- a/java/org/apache/catalina/tribes/jmx/JmxRegistry.java
+++ b/java/org/apache/catalina/tribes/jmx/JmxRegistry.java
@@ -88,12 +88,16 @@ public class JmxRegistry {
}
/**
- * Remove the JMX registry for the given channel.
+ * Remove the JMX registry for the given channel. Does nothing if the channel or its name is
+ * {@code null}.
*
* @param channel the channel
* @param clear whether to clear the MBeans
*/
public static void removeRegistry(Channel channel, boolean clear) {
+ if (channel == null || channel.getName() == null) {
+ return;
+ }
JmxRegistry registry = registryCache.get(channel.getName());
if (registry == null) {
return;
diff --git a/test/org/apache/catalina/tribes/jmx/TestJmxRegistry.java b/test/org/apache/catalina/tribes/jmx/TestJmxRegistry.java
index fa44f85e45..bc61f8a0b5 100644
--- a/test/org/apache/catalina/tribes/jmx/TestJmxRegistry.java
+++ b/test/org/apache/catalina/tribes/jmx/TestJmxRegistry.java
@@ -100,4 +100,18 @@ public class TestJmxRegistry {
JmxRegistry.removeRegistry(channel, true);
}
}
+
+ /*
+ * Removing the registry of a channel that never had one, or that cannot have one, must degrade
+ * to a no-op, as getRegistry does, rather than throw.
+ */
+ @Test
+ public void testRemoveRegistryNullGuards() throws Exception {
+ JmxRegistry.removeRegistry(null, false);
+ JmxRegistry.removeRegistry(null, true);
+
+ GroupChannel channel = new GroupChannel();
+ JmxRegistry.removeRegistry(channel, false);
+ JmxRegistry.removeRegistry(channel, true);
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
(tomcat) 05/20: Guard the sender recovery trigger against a concurrent stop
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 03c16b4c50e77eb835ef89f0b9cd23bdd2234d6a
Author: opencode <re...@apache.org>
AuthorDate: Sun Oct 11 10:34:59 2026 +0200
Guard the sender recovery trigger against a concurrent stop
The multicast sender triggered a service recovery once it had hit
recoveryCounter consecutive send failures, without re-checking the
doRunSender flag that ReceiverThread consults before the same action.
Since the recovery thread always recreates the socket and the
receiver and sender threads, the last send failure of a closing
socket could schedule a recovery that brought the membership back
after McastService.stop() had already discarded the implementation.
Nothing then held a reference to stop that revived implementation, so
it kept pinging with the old unique id, leaving peers unable to
expire the departed member, and leaked the socket and two threads.
The counter increment and the recovery call now happen only while
doRunSender is still set, exactly as in the receiver. Recovery on a
running service is unchanged.
---
java/org/apache/catalina/tribes/membership/McastServiceImpl.java | 8 +++++---
1 file changed, 5 insertions(+), 3 deletions(-)
diff --git a/java/org/apache/catalina/tribes/membership/McastServiceImpl.java b/java/org/apache/catalina/tribes/membership/McastServiceImpl.java
index 1eec1a731c..d5c538026f 100644
--- a/java/org/apache/catalina/tribes/membership/McastServiceImpl.java
+++ b/java/org/apache/catalina/tribes/membership/McastServiceImpl.java
@@ -697,9 +697,11 @@ public class McastServiceImpl extends MembershipProviderBase {
} else {
log.debug(sm.getString("mcastServiceImpl.send.failed"), e);
}
- if ((++errorCounter) >= recoveryCounter) {
- errorCounter = 0;
- RecoveryThread.recover(McastServiceImpl.this);
+ if (doRunSender) {
+ if ((++errorCounter) >= recoveryCounter) {
+ errorCounter = 0;
+ RecoveryThread.recover(McastServiceImpl.this);
+ }
}
}
try {
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
(tomcat) 09/20: Reject non-positive membership timing properties at start
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 815d559223f59234deda4f4ca049e97c960ff706
Author: opencode <re...@apache.org>
AuthorDate: Sun Oct 11 11:24:49 2026 +0200
Reject non-positive membership timing properties at start
The timing properties of the membership services were stored as they
were configured, which let a non positive value take effect. A non
positive mcast frequency made the multicast sender send pings in a
tight loop, as the exception that a non positive sleep raises is
swallowed around the send cycle, and a non positive member drop time
expired every member on every cycle, churning memberDisappeared and
memberAdded events. The static membership provider showed the same
behaviour with its ping interval and expiration time.
Both services now validate the parsed values where the configuration
converges, the multicast service when it builds its implementation and
the static provider when it initialises, and reject them with an
illegal argument exception naming the property and its value. The
checks run before any socket or thread exists and cover both the
setters and a replaced set of properties, as the setters only write
the properties that these points read. The check of the ping interval
moved to the top of the provider initialisation, next to the one of
the expiration time, so that neither can leave a channel listener
registered behind a rejected configuration.
Add tests that a non positive value is rejected by each service before
anything starts and that the static service still starts and stops
with the default configuration.
---
.../tribes/membership/LocalStrings.properties | 2 +
.../catalina/tribes/membership/McastService.java | 14 +++-
.../membership/StaticMembershipProvider.java | 12 +++-
.../tribes/membership/TestMcastService.java | 37 +++++++++++
.../membership/TestStaticMembershipService.java | 77 ++++++++++++++++++++++
5 files changed, 138 insertions(+), 4 deletions(-)
diff --git a/java/org/apache/catalina/tribes/membership/LocalStrings.properties b/java/org/apache/catalina/tribes/membership/LocalStrings.properties
index 9a2af3489c..97d1dbdb5b 100644
--- a/java/org/apache/catalina/tribes/membership/LocalStrings.properties
+++ b/java/org/apache/catalina/tribes/membership/LocalStrings.properties
@@ -26,6 +26,8 @@ mcastService.exceed.maxPacketSize=Packet length [{0}] exceeds max packet size of
mcastService.missing.property=McastService:Required property [{0}] is missing.
mcastService.noStart=Multicast send is not started or enabled.
+membershipService.invalidTimingProperty=The timing property [{0}] is set to [{1}]. It must be greater than zero.
+
mcastServiceImpl.bind=Attempting to bind the multicast socket to [{0}:{1}]
mcastServiceImpl.bind.failed=Binding to multicast address, failed. Binding to port only.
mcastServiceImpl.error.receiving=Error receiving mcast package. Sleeping 500ms
diff --git a/java/org/apache/catalina/tribes/membership/McastService.java b/java/org/apache/catalina/tribes/membership/McastService.java
index 0a05ed2f4d..900ed90271 100644
--- a/java/org/apache/catalina/tribes/membership/McastService.java
+++ b/java/org/apache/catalina/tribes/membership/McastService.java
@@ -383,8 +383,18 @@ public class McastService extends MembershipServiceBase implements MessageListen
}
}
- impl = new McastServiceImpl(localMember, Long.parseLong(properties.getProperty("mcastFrequency")),
- Long.parseLong(properties.getProperty("memberDropTime")),
+ long frequency = Long.parseLong(properties.getProperty("mcastFrequency"));
+ if (frequency <= 0) {
+ throw new IllegalArgumentException(
+ sm.getString("membershipService.invalidTimingProperty", "mcastFrequency", Long.valueOf(frequency)));
+ }
+ long dropTime = Long.parseLong(properties.getProperty("memberDropTime"));
+ if (dropTime <= 0) {
+ throw new IllegalArgumentException(
+ sm.getString("membershipService.invalidTimingProperty", "memberDropTime", Long.valueOf(dropTime)));
+ }
+
+ impl = new McastServiceImpl(localMember, frequency, dropTime,
Integer.parseInt(properties.getProperty("mcastPort")), bind,
java.net.InetAddress.getByName(properties.getProperty("mcastAddress")), ttl, soTimeout, this, this,
Boolean.parseBoolean(properties.getProperty("localLoopbackDisabled")));
diff --git a/java/org/apache/catalina/tribes/membership/StaticMembershipProvider.java b/java/org/apache/catalina/tribes/membership/StaticMembershipProvider.java
index 3e94478057..d76b0953f1 100644
--- a/java/org/apache/catalina/tribes/membership/StaticMembershipProvider.java
+++ b/java/org/apache/catalina/tribes/membership/StaticMembershipProvider.java
@@ -122,6 +122,16 @@ public class StaticMembershipProvider extends MembershipProviderBase
// All the default properties have been set by StaticMembershipService.setDefaults, so no need to check
String expirationTimeStr = properties.getProperty("expirationTime");
this.expirationTime = Long.parseLong(expirationTimeStr);
+ if (expirationTime <= 0) {
+ throw new IllegalArgumentException(sm.getString("membershipService.invalidTimingProperty", "expirationTime",
+ Long.valueOf(expirationTime)));
+ }
+ String pingIntervalStr = properties.getProperty("pingInterval");
+ this.pingInterval = Long.parseLong(pingIntervalStr);
+ if (pingInterval <= 0) {
+ throw new IllegalArgumentException(sm.getString("membershipService.invalidTimingProperty", "pingInterval",
+ Long.valueOf(pingInterval)));
+ }
String connectTimeoutStr = properties.getProperty("connectTimeout");
this.connectTimeout = Integer.parseInt(connectTimeoutStr);
String rpcTimeouStr = properties.getProperty("rpcTimeout");
@@ -133,8 +143,6 @@ public class StaticMembershipProvider extends MembershipProviderBase
this.channel.addChannelListener(this);
String useThreadStr = properties.getProperty("useThread");
this.useThread = Boolean.parseBoolean(useThreadStr);
- String pingIntervalStr = properties.getProperty("pingInterval");
- this.pingInterval = Long.parseLong(pingIntervalStr);
}
@Override
diff --git a/test/org/apache/catalina/tribes/membership/TestMcastService.java b/test/org/apache/catalina/tribes/membership/TestMcastService.java
index 77e43af28c..fa0144b841 100644
--- a/test/org/apache/catalina/tribes/membership/TestMcastService.java
+++ b/test/org/apache/catalina/tribes/membership/TestMcastService.java
@@ -123,6 +123,43 @@ public class TestMcastService {
Assert.assertEquals(0, service.getTtl());
}
+ /*
+ * A non-positive send frequency makes the sender spin sending pings without a pause, and a
+ * non-positive drop time expires every member on every cycle. Such configuration must be
+ * rejected when the service starts, before any socket or thread is created.
+ */
+ @Test
+ public void testStartRejectsNonPositiveTimingProperties() throws Exception {
+ String[] invalid = { "mcastFrequency=0", "mcastFrequency=-1", "memberDropTime=0", "memberDropTime=-5" };
+ for (String entry : invalid) {
+ String[] property = entry.split("=");
+ Properties props = new Properties();
+ props.setProperty("mcastAddress", "228.0.0.4");
+ props.setProperty("mcastPort", "45564");
+ props.setProperty("tcpListenHost", "127.0.0.1");
+ props.setProperty("tcpListenPort", "4000");
+ props.setProperty("tcpSecurePort", "0");
+ props.setProperty("udpListenPort", "0");
+ props.setProperty("mcastFrequency", "500");
+ props.setProperty("memberDropTime", "3000");
+ props.setProperty(property[0], property[1]);
+
+ GroupChannel channel = new GroupChannel();
+ channel.setName("TestMcastService-" + System.nanoTime());
+
+ McastService service = new McastService();
+ service.setChannel(channel);
+ service.setProperties(props);
+ try {
+ service.start(Channel.MBR_TX_SEQ);
+ Assert.fail("start() must reject a non-positive " + property[0]);
+ } catch (IllegalArgumentException expected) {
+ // The configuration must be rejected before anything is started
+ }
+ Assert.assertNull(service.impl);
+ }
+ }
+
/*
* The 65535 byte protocol limit of an IP packet includes the IP and UDP headers, so a
* broadcast that fits within it can still exceed what the socket is able to send. Such a
diff --git a/test/org/apache/catalina/tribes/membership/TestStaticMembershipService.java b/test/org/apache/catalina/tribes/membership/TestStaticMembershipService.java
new file mode 100644
index 0000000000..c2b3992f18
--- /dev/null
+++ b/test/org/apache/catalina/tribes/membership/TestStaticMembershipService.java
@@ -0,0 +1,77 @@
+/*
+ * 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.membership;
+
+import java.util.Properties;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import org.apache.catalina.tribes.Channel;
+import org.apache.catalina.tribes.group.GroupChannel;
+
+public class TestStaticMembershipService {
+
+ /*
+ * A non-positive ping interval makes the ping thread spin without a pause, and a non-positive
+ * expiration time expires every member on every cycle. Such configuration must be rejected
+ * when the service initializes its provider.
+ */
+ @Test
+ public void testStartRejectsNonPositiveTimingProperties() throws Exception {
+ String[] invalid = { "pingInterval=0", "pingInterval=-1", "expirationTime=0", "expirationTime=-5" };
+ for (String entry : invalid) {
+ String[] property = entry.split("=");
+ Properties props = new Properties();
+ props.setProperty(property[0], property[1]);
+
+ GroupChannel channel = new GroupChannel();
+ channel.setName("TestStaticMembershipService-" + System.nanoTime());
+
+ StaticMembershipService service = new StaticMembershipService();
+ service.setChannel(channel);
+ service.setProperties(props);
+ // The local member must be part of the static member list
+ service.addStaticMember(new StaticMember("127.0.0.1", 4001, -1));
+ service.setLocalMemberProperties("127.0.0.1", 4001, 0, 0);
+ try {
+ service.start(Channel.MBR_TX_SEQ);
+ Assert.fail("start() must reject a non-positive " + property[0]);
+ } catch (IllegalArgumentException expected) {
+ // The configuration must be rejected before the provider is built
+ }
+ }
+ }
+
+ /*
+ * A valid configuration, as setDefaults provides it, must still start and stop cleanly.
+ */
+ @Test
+ public void testValidTimingPropertiesStartAndStop() throws Exception {
+ GroupChannel channel = new GroupChannel();
+ channel.setName("TestStaticMembershipService-" + System.nanoTime());
+
+ StaticMembershipService service = new StaticMembershipService();
+ service.setChannel(channel);
+ service.addStaticMember(new StaticMember("127.0.0.1", 4002, -1));
+ service.setLocalMemberProperties("127.0.0.1", 4002, 0, 0);
+
+ int level = Channel.MBR_RX_SEQ | Channel.MBR_TX_SEQ;
+ service.start(level);
+ service.stop(level);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
(tomcat) 15/20: Clamp the backup index to the member snapshot in publishEntryInfo
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 0de06540fa0b547f1175e998429c5e4899fbb932
Author: opencode <re...@apache.org>
AuthorDate: Sun Oct 11 13:09:09 2026 +0200
Clamp the backup index to the member snapshot in publishEntryInfo
The backup node selection snapshots the members with getMapMembers
and then computes the starting index with getNextBackupIndex, which
clamps against the current member count rather than the snapshot. A
member added between the two steps lets the returned index point
beyond the end of the snapshot, and the first array access in the
selection loop then threw an array index out of bounds exception.
That exception was not caught by put(), which only handles channel
exceptions, so it escaped before the entry was stored in the inner
map. With the BackupManager, whose session storage is the replicated
map, this lost a session that was being created while a node joined
the cluster.
The selection now falls back to the start of the snapshot when the
index is out of its range, mirroring the existing guard in
getNextBackupNode. Add a unit test that pins the diverging snapshot
and index and drives the selection to completion.
---
.../catalina/tribes/tipis/LazyReplicatedMap.java | 8 ++-
.../tribes/tipis/TestLazyReplicatedMap.java | 61 ++++++++++++++++++++++
2 files changed, 68 insertions(+), 1 deletion(-)
diff --git a/java/org/apache/catalina/tribes/tipis/LazyReplicatedMap.java b/java/org/apache/catalina/tribes/tipis/LazyReplicatedMap.java
index 8780a50564..b1b9bd207f 100644
--- a/java/org/apache/catalina/tribes/tipis/LazyReplicatedMap.java
+++ b/java/org/apache/catalina/tribes/tipis/LazyReplicatedMap.java
@@ -158,13 +158,19 @@ public class LazyReplicatedMap<K, V> extends AbstractReplicatedMap<K,V> {
}
Member[] members = getMapMembers();
int firstIdx = getNextBackupIndex();
- int nextIdx = firstIdx;
Member[] backup = new Member[0];
// there are no backups
if (members.length == 0 || firstIdx == -1) {
return backup;
}
+ // A member may have been added between the snapshot of the members and the computation of
+ // the index, which then points beyond the end of the snapshot. Fall back to its start, as
+ // getNextBackupNode does.
+ if (firstIdx >= members.length) {
+ firstIdx = 0;
+ }
+ int nextIdx = firstIdx;
boolean success = false;
do {
diff --git a/test/org/apache/catalina/tribes/tipis/TestLazyReplicatedMap.java b/test/org/apache/catalina/tribes/tipis/TestLazyReplicatedMap.java
new file mode 100644
index 0000000000..f6cc37fec2
--- /dev/null
+++ b/test/org/apache/catalina/tribes/tipis/TestLazyReplicatedMap.java
@@ -0,0 +1,61 @@
+/*
+ * 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.tipis;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.group.GroupChannel;
+import org.apache.catalina.tribes.membership.MemberImpl;
+
+public class TestLazyReplicatedMap {
+
+ /*
+ * The backup node selection snapshots the members and computes an index against the current
+ * members in two steps. A member added in between lets the index point behind the end of the
+ * snapshot, which aborted the publication with an array index out of bounds exception, losing
+ * the entry that was being stored. The index must fall back to the start of the snapshot.
+ */
+ @Test
+ public void testBackupIndexBeyondMemberSnapshotFallsBack() throws Exception {
+ Member member = new MemberImpl("localhost", 4007, -1);
+
+ // A member added between the two steps of the selection, shown directly: one member in
+ // the snapshot, an index computed for a member list that has meanwhile grown
+ LazyReplicatedMap<String,String> map = new LazyReplicatedMap<String,String>(
+ (key, value) -> {
+ // Not made primary here
+ }, new GroupChannel(), 1000, "TestLazyReplicatedMap", null) {
+
+ @Override
+ public Member[] getMapMembers() {
+ return new Member[] { member };
+ }
+
+ @Override
+ public int getNextBackupIndex() {
+ return 1;
+ }
+ };
+
+ // The channel is not started, so the publication to the backup fails as a channel
+ // exception that the selection contains, while the entry survives to be stored locally
+ Member[] backup = map.publishEntryInfo("key", "value");
+ Assert.assertEquals(0, backup.length);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
(tomcat) 07/20: Report the configured mcast so timeout and TTL through JMX
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit f3d134171ffd729fd40a1e97bc2aeff3b2915f71
Author: opencode <re...@apache.org>
AuthorDate: Sun Oct 11 10:52:07 2026 +0200
Report the configured mcast so timeout and TTL through JMX
The getSoTimeout() and getTtl() attributes of the membership service
MBean returned private fields that only the matching setters wrote,
while start() configures the multicast service from the mcastSoTimeout
and mcastTTL properties directly. A deployment that configures the
multicast membership through its properties, rather than through the
setters, therefore ran with a socket timeout and TTL that JMX reported
as the field default of zero.
The two attributes now read the properties, as every other
property-backed attribute of the service does, with the previous zero
default kept for an unset property, which also makes them report what
a subsequent start would apply rather than a setter value that a
replaced set of properties dropped on the floor. The fields behind the
getters are no longer used and were removed.
Add tests for the property based configuration, the setters, the
default of an unset property, and the precedence of a replaced set of
properties over an earlier setter value.
---
.../catalina/tribes/membership/McastService.java | 16 ++++---
.../tribes/membership/TestMcastService.java | 50 ++++++++++++++++++++++
2 files changed, 60 insertions(+), 6 deletions(-)
diff --git a/java/org/apache/catalina/tribes/membership/McastService.java b/java/org/apache/catalina/tribes/membership/McastService.java
index cda9f12608..0672086983 100644
--- a/java/org/apache/catalina/tribes/membership/McastService.java
+++ b/java/org/apache/catalina/tribes/membership/McastService.java
@@ -64,8 +64,6 @@ public class McastService extends MembershipServiceBase implements MessageListen
* The local member
*/
protected MemberImpl localMember;
- private int mcastSoTimeout;
- private int mcastTTL;
/**
* The payload for multicast messages.
@@ -482,7 +480,11 @@ public class McastService extends MembershipServiceBase implements MessageListen
@Override
public int getSoTimeout() {
- return mcastSoTimeout;
+ String p = properties.getProperty("mcastSoTimeout");
+ if (p != null) {
+ return Integer.parseInt(p);
+ }
+ return 0;
}
/**
@@ -490,13 +492,16 @@ public class McastService extends MembershipServiceBase implements MessageListen
* @param mcastSoTimeout the timeout in milliseconds
*/
public void setSoTimeout(int mcastSoTimeout) {
- this.mcastSoTimeout = mcastSoTimeout;
properties.setProperty("mcastSoTimeout", String.valueOf(mcastSoTimeout));
}
@Override
public int getTtl() {
- return mcastTTL;
+ String p = properties.getProperty("mcastTTL");
+ if (p != null) {
+ return Integer.parseInt(p);
+ }
+ return 0;
}
/**
@@ -517,7 +522,6 @@ public class McastService extends MembershipServiceBase implements MessageListen
* @param mcastTTL the TTL value
*/
public void setTtl(int mcastTTL) {
- this.mcastTTL = mcastTTL;
properties.setProperty("mcastTTL", String.valueOf(mcastTTL));
}
diff --git a/test/org/apache/catalina/tribes/membership/TestMcastService.java b/test/org/apache/catalina/tribes/membership/TestMcastService.java
index ad5b981db5..431e6fae9a 100644
--- a/test/org/apache/catalina/tribes/membership/TestMcastService.java
+++ b/test/org/apache/catalina/tribes/membership/TestMcastService.java
@@ -69,4 +69,54 @@ public class TestMcastService {
Assert.assertNull("the membership implementation must be released on stop", service.impl);
}
+
+ /*
+ * The soTimeout and ttl reported through MBean must reflect the configuration, whether it is
+ * supplied through the properties file or through the setters. Previously the getters read
+ * private fields that only the setters wrote, so a property-based configuration that the socket
+ * did apply was reported as the default 0.
+ */
+ @Test
+ public void testSoTimeoutAndTtlReflectProperties() throws Exception {
+ Properties properties = new Properties();
+ properties.setProperty("mcastAddress", "228.0.0.4");
+ properties.setProperty("mcastPort", "45564");
+ properties.setProperty("memberDropTime", "3000");
+ properties.setProperty("mcastFrequency", "500");
+ properties.setProperty("tcpListenHost", "127.0.0.1");
+ properties.setProperty("tcpListenPort", "4000");
+ properties.setProperty("mcastSoTimeout", "1234");
+ properties.setProperty("mcastTTL", "5");
+
+ McastService service = new McastService();
+ service.setProperties(properties);
+
+ Assert.assertEquals(1234, service.getSoTimeout());
+ Assert.assertEquals(5, service.getTtl());
+
+ // The setters keep overriding, with the setter value taking precedence
+ service.setSoTimeout(7);
+ service.setTtl(9);
+ Assert.assertEquals(7, service.getSoTimeout());
+ Assert.assertEquals(9, service.getTtl());
+
+ // A later setProperties replaces the whole configuration, and with it the setter values,
+ // as that is what a subsequent start() would apply
+ properties.remove("mcastSoTimeout");
+ properties.remove("mcastTTL");
+ service.setProperties(properties);
+ Assert.assertEquals(0, service.getSoTimeout());
+ Assert.assertEquals(0, service.getTtl());
+ }
+
+ /*
+ * When neither the properties nor a setter provides a value, the reported soTimeout and ttl
+ * default to 0, as they did before.
+ */
+ @Test
+ public void testSoTimeoutAndTtlDefaultToZero() throws Exception {
+ McastService service = new McastService();
+ Assert.assertEquals(0, service.getSoTimeout());
+ Assert.assertEquals(0, service.getTtl());
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
(tomcat) 01/20: Validate package size field in XByteBuffer package framing
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 8393a3cbf1d4642f0fc395ed65eb3d3ba129a231
Author: opencode <re...@apache.org>
AuthorDate: Sun Oct 11 10:01:18 2026 +0200
Validate package size field in XByteBuffer package framing
The package size read from the wire was used without validation. A
negative size for the second or a later package could make the footer
check land on the footer of the preceding package, resetting the loop
start to its previous value, so countPackages() never terminated and
permanently spun one of the replication receiver workers or the
multicast receiver thread. A negative or overflowingly large size for
the first package instead threw an ArrayIndexOutOfBoundsException when
scanning for the package footer.
Sizes are now rejected, consistent with a footer mismatch, when they
are negative or when the package they describe cannot fit within the
buffered data. The length check is done with long arithmetic so that a
large size cannot overflow and wrap around to a small or negative
value, and the offset guard of firstIndexOf() now also rejects negative
offsets.
Add tests for the livelock and exception inputs and for valid package
counting.
---
.../org/apache/catalina/tribes/io/XByteBuffer.java | 14 ++---
.../apache/catalina/tribes/io/TestXByteBuffer.java | 63 ++++++++++++++++++++++
2 files changed, 71 insertions(+), 6 deletions(-)
diff --git a/java/org/apache/catalina/tribes/io/XByteBuffer.java b/java/org/apache/catalina/tribes/io/XByteBuffer.java
index 10e8e8e27e..2db187ecc2 100644
--- a/java/org/apache/catalina/tribes/io/XByteBuffer.java
+++ b/java/org/apache/catalina/tribes/io/XByteBuffer.java
@@ -371,12 +371,14 @@ public class XByteBuffer implements Serializable {
}
// next 4 bytes are the size of the package
int size = toInt(buf, pos);
- // now the total buffer has to be long enough to hold
- // START_DATA.length+4+size+END_DATA.length
- pos = start + START_DATA.length + 4 + size;
- if ((pos + END_DATA.length) > bufSize) {
+ // the size must not be negative and the total buffer has to be
+ // long enough to hold START_DATA.length+4+size+END_DATA.length;
+ // use long arithmetic so that a large size cannot overflow and
+ // wrap around to a negative or small value
+ if (size < 0 || (long) start + START_DATA.length + 4 + size + END_DATA.length > bufSize) {
break;
}
+ pos = start + START_DATA.length + 4 + size;
// and finally check the footer of the package END_DATA
int newpos = firstIndexOf(buf, pos, END_DATA);
// mismatch, there is no package
@@ -635,8 +637,8 @@ public class XByteBuffer implements Serializable {
if (find.length == 0) {
return result;
}
- if (srcOff >= src.length) {
- throw new ArrayIndexOutOfBoundsException();
+ if (srcOff < 0 || srcOff >= src.length) {
+ throw new ArrayIndexOutOfBoundsException(srcOff);
}
boolean found = false;
int srclen = src.length;
diff --git a/test/org/apache/catalina/tribes/io/TestXByteBuffer.java b/test/org/apache/catalina/tribes/io/TestXByteBuffer.java
index cf7ff86cc3..0487c75950 100644
--- a/test/org/apache/catalina/tribes/io/TestXByteBuffer.java
+++ b/test/org/apache/catalina/tribes/io/TestXByteBuffer.java
@@ -38,4 +38,67 @@ public class TestXByteBuffer {
assertThat(obj, instanceOf(String.class));
Assert.assertEquals(test, obj);
}
+
+ private static byte[] packageOf(int... message) {
+ XByteBuffer msg = new XByteBuffer(new byte[0], false);
+ for (int b : message) {
+ msg.append(new byte[] { (byte) b }, 0, 1);
+ }
+ return XByteBuffer.createDataPackage(new ChannelData(new byte[] { 1, 2 }, msg, System.currentTimeMillis()));
+ }
+
+ private static XByteBuffer validPackageBuffer() {
+ return new XByteBuffer(packageOf(3), false);
+ }
+
+ private static byte[] packageStub(int size) {
+ // A header, a size field and three filler bytes, the minimum that
+ // countPackages() inspects before the footer check
+ byte[] stub = new byte[14];
+ System.arraycopy(packageOf(2), 0, stub, 0, 7);
+ XByteBuffer.toBytes(size, stub, 7);
+ stub[11] = 'x';
+ stub[12] = 'y';
+ stub[13] = 'z';
+ return stub;
+ }
+
+ @Test
+ public void testCountPackagesValid() throws Exception {
+ XByteBuffer buf = validPackageBuffer();
+ Assert.assertEquals(1, buf.countPackages());
+ Assert.assertEquals(1, buf.countPackages(true));
+ byte[] pkg = packageOf(2, 3);
+ buf.append(pkg, 0, pkg.length);
+ Assert.assertEquals(2, buf.countPackages());
+ Assert.assertEquals(1, buf.countPackages(true));
+ }
+
+ // A size field of -18 makes the footer check land on the footer of the
+ // preceding package. This used to reset the loop start to its previous
+ // value, so countPackages() never terminated.
+ @Test(timeout = 30000)
+ public void testCountPackagesNegativeSizeAfterValidPackage() throws Exception {
+ XByteBuffer buf = validPackageBuffer();
+ byte[] stub = packageStub(-18);
+ buf.append(stub, 0, stub.length);
+ Assert.assertEquals(1, buf.countPackages());
+ Assert.assertEquals(1, buf.countPackages(true));
+ }
+
+ @Test
+ public void testCountPackagesNegativeSize() throws Exception {
+ XByteBuffer buf = new XByteBuffer(4096, false);
+ byte[] stub = packageStub(-18);
+ buf.append(stub, 0, stub.length);
+ Assert.assertEquals(0, buf.countPackages());
+ }
+
+ @Test
+ public void testCountPackagesHugeSize() throws Exception {
+ XByteBuffer buf = new XByteBuffer(4096, false);
+ byte[] stub = packageStub(Integer.MAX_VALUE);
+ buf.append(stub, 0, stub.length);
+ Assert.assertEquals(0, buf.countPackages());
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
(tomcat) 02/20: Make the buffer pool size accounting atomic
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit c9ede40486f77ce5630a56abe5f78c5b6e263d27
Author: opencode <re...@apache.org>
AuthorDate: Sun Oct 11 10:13:38 2026 +0200
Make the buffer pool size accounting atomic
The pool size limit was enforced with a check against the current
total followed by a separate increment, so concurrent returns could
all pass the check against the same stale total and admit more bytes
between them than the configured maximum. Clearing the pool was not
atomic either. A return that interleaved between the queue clear and
the size reset left the accounting permanently skewed, a buffer kept
in the queue without being counted, or counted without being kept,
biasing every later admission decision over or under the limit.
The limit is now enforced with a compare-and-set loop, making the size
never exceed the maximum under concurrent returns, and the clear now
removes each buffer from the queue and its capacity from the total as
one paired operation, so the size again matches the buffers that the
queue holds once the pool is quiescent. This also removes a related
skew where a subtraction landing after the reset could drive the size
negative.
Add tests that race returns against the limit and against a concurrent
clear, both of which fail reliably on the previous code.
---
java/org/apache/catalina/tribes/io/BufferPool.java | 27 +++-
.../apache/catalina/tribes/io/TestBufferPool.java | 153 +++++++++++++++++++++
2 files changed, 174 insertions(+), 6 deletions(-)
diff --git a/java/org/apache/catalina/tribes/io/BufferPool.java b/java/org/apache/catalina/tribes/io/BufferPool.java
index fae2798cb1..7b4656c5b9 100644
--- a/java/org/apache/catalina/tribes/io/BufferPool.java
+++ b/java/org/apache/catalina/tribes/io/BufferPool.java
@@ -90,16 +90,31 @@ public class BufferPool {
* @param buffer the buffer to return
*/
public void returnBuffer(XByteBuffer buffer) {
- if ((size.get() + buffer.getCapacity()) <= maxSize) {
- size.addAndGet(buffer.getCapacity());
- queue.offer(buffer);
- }
+ // the pool size limit is enforced with a compare-and-set loop so that
+ // concurrent returns cannot all pass the limit check against the same
+ // stale total and admit more than maxSize bytes between them
+ int capacity = buffer.getCapacity();
+ int current;
+ do {
+ current = size.get();
+ if ((long) current + capacity > maxSize) {
+ return;
+ }
+ } while (!size.compareAndSet(current, current + capacity));
+ queue.offer(buffer);
}
/** Clears all buffers from the pool. */
public void clear() {
- queue.clear();
- size.set(0);
+ // drain the queue buffer by buffer, removing each capacity from the
+ // size total as the buffer leaves the queue; setting the size to zero
+ // separately would race with concurrent returns and skew the
+ // accounting permanently
+ XByteBuffer buffer = queue.poll();
+ while (buffer != null) {
+ size.addAndGet(-buffer.getCapacity());
+ buffer = queue.poll();
+ }
}
/** Maximum pool size in bytes. */
diff --git a/test/org/apache/catalina/tribes/io/TestBufferPool.java b/test/org/apache/catalina/tribes/io/TestBufferPool.java
new file mode 100644
index 0000000000..2ed916c71d
--- /dev/null
+++ b/test/org/apache/catalina/tribes/io/TestBufferPool.java
@@ -0,0 +1,153 @@
+/*
+ * 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.io;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestBufferPool {
+
+ private static final int MAX = 512 * 1024;
+
+ /*
+ * Concurrent returns must not all pass the limit check against the same stale total and admit
+ * more than maxSize bytes between them. The pool is parked just below the limit and repeatedly
+ * offered to the racing threads, so the limit check is exercised over and over on the edge.
+ */
+ @Test
+ public void testConcurrentReturnsRespectMaxSize() throws Exception {
+ BufferPool pool = BufferPool.getBufferPool();
+ int original = pool.getMaxSize();
+ try {
+ pool.clear();
+ pool.setMaxSize(MAX);
+ final AtomicBoolean go = new AtomicBoolean();
+ final AtomicBoolean stop = new AtomicBoolean();
+ List<Thread> threads = new ArrayList<>();
+ try {
+ for (int t = 0; t < 4; t++) {
+ final XByteBuffer[] ring = new XByteBuffer[32];
+ for (int i = 0; i < ring.length; i++) {
+ ring[i] = new XByteBuffer(64 * 1024, false);
+ }
+ Thread thread = new Thread(() -> {
+ int index = 0;
+ while (!stop.get()) {
+ while (!go.get() && !stop.get()) {
+ Thread.onSpinWait();
+ }
+ while (go.get() && !stop.get()) {
+ pool.returnBuffer(ring[index++ & 31]);
+ }
+ }
+ });
+ thread.setDaemon(true);
+ threads.add(thread);
+ thread.start();
+ }
+ int overshoot = 0;
+ int trials = 750;
+ for (int trial = 0; trial < trials; trial++) {
+ drain(pool);
+ for (int i = 0; i < MAX / (64 * 1024) - 1; i++) {
+ pool.returnBuffer(new XByteBuffer(64 * 1024, false));
+ }
+ go.set(true);
+ Thread.sleep(0, 200000);
+ go.set(false);
+ if (pool.size.get() > MAX) {
+ overshoot++;
+ }
+ }
+ Assert.assertEquals("pool size exceeded maxSize in " + overshoot + " of the concurrent trials", 0,
+ overshoot);
+ } finally {
+ stop.set(true);
+ for (Thread thread : threads) {
+ thread.join();
+ }
+ }
+ } finally {
+ pool.setMaxSize(original);
+ pool.clear();
+ }
+ }
+
+ private static void drain(BufferPool pool) {
+ while (!pool.queue.isEmpty()) {
+ pool.getBuffer(1, false);
+ }
+ }
+
+ /*
+ * Clearing the pool while buffers are returned must not skew the size accounting. Once all
+ * activity has stopped, the size must again match the buffers that the queue actually holds.
+ */
+ @Test
+ public void testClearKeepsAccountingConsistent() throws Exception {
+ BufferPool pool = BufferPool.getBufferPool();
+ int original = pool.getMaxSize();
+ try {
+ pool.clear();
+ pool.setMaxSize(MAX);
+ AtomicBoolean stop = new AtomicBoolean();
+ Thread returns = new Thread(() -> {
+ while (!stop.get()) {
+ pool.returnBuffer(new XByteBuffer(1024, false));
+ }
+ });
+ returns.setDaemon(true);
+ Thread[] clears = new Thread[2];
+ try {
+ returns.start();
+ for (int t = 0; t < clears.length; t++) {
+ clears[t] = new Thread(() -> {
+ while (!stop.get()) {
+ pool.clear();
+ }
+ });
+ clears[t].setDaemon(true);
+ clears[t].start();
+ }
+ Thread.sleep(250);
+ } finally {
+ stop.set(true);
+ returns.join();
+ for (Thread thread : clears) {
+ thread.join();
+ }
+ }
+ // Once all activity has stopped, the size must again match the
+ // buffers that the queue actually holds
+ long queued = 0;
+ for (XByteBuffer buffer : pool.queue) {
+ queued += buffer.getCapacity();
+ }
+ Assert.assertEquals("pool size must match the capacity of the queued buffers", queued, pool.size.get());
+ pool.clear();
+ Assert.assertEquals(0, pool.size.get());
+ Assert.assertTrue(pool.queue.isEmpty());
+ } finally {
+ pool.setMaxSize(original);
+ pool.clear();
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
(tomcat) 13/20: Hand out a copy of the static members list
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit ab90082aaa7e7d472d2c9a70148c26812052acfc
Author: opencode <re...@apache.org>
AuthorDate: Sun Oct 11 12:34:11 2026 +0200
Hand out a copy of the static members list
The accessor of the static members returned the internal list that the
membership provider iterates from its ping thread and at start and
stop. A caller that mutated the returned list at runtime raced that
iteration of an unsynchronized array list, with the member list
occasionally read in an inconsistent state and members silently
missing from pings and start notifications.
The accessor now hands out a copy, and its contract says so, leaving
the provider with the configured list, which only the service itself
still touches. The signature keeps the array list type, so callers
compiled against the accessor are unaffected, while the read only
callers in this repository keep working against the copy.
Add tests pinning the contract, an accessor that returns a different
list each call and mutations of the returned list that leave the
provider list untouched.
---
.../tribes/membership/StaticMembershipService.java | 7 ++++---
.../membership/TestStaticMembershipService.java | 22 ++++++++++++++++++++++
2 files changed, 26 insertions(+), 3 deletions(-)
diff --git a/java/org/apache/catalina/tribes/membership/StaticMembershipService.java b/java/org/apache/catalina/tribes/membership/StaticMembershipService.java
index b4d9d3101e..2d4f8c399e 100644
--- a/java/org/apache/catalina/tribes/membership/StaticMembershipService.java
+++ b/java/org/apache/catalina/tribes/membership/StaticMembershipService.java
@@ -164,11 +164,12 @@ public class StaticMembershipService extends MembershipServiceBase implements St
}
/**
- * Returns the list of static members.
- * @return the static members list
+ * Returns a copy of the list of static members. Mutating the returned list has no effect on
+ * this service. The provider uses the configured list directly while it runs.
+ * @return a copy of the static members list
*/
public ArrayList<StaticMember> getStaticMembers() {
- return staticMembers;
+ return new ArrayList<>(staticMembers);
}
/**
diff --git a/test/org/apache/catalina/tribes/membership/TestStaticMembershipService.java b/test/org/apache/catalina/tribes/membership/TestStaticMembershipService.java
index ddafa78835..71d35f5c5c 100644
--- a/test/org/apache/catalina/tribes/membership/TestStaticMembershipService.java
+++ b/test/org/apache/catalina/tribes/membership/TestStaticMembershipService.java
@@ -16,6 +16,7 @@
*/
package org.apache.catalina.tribes.membership;
+import java.util.ArrayList;
import java.util.Properties;
import org.junit.Assert;
@@ -120,4 +121,25 @@ public class TestStaticMembershipService {
service.stop(level);
}
}
+
+ /*
+ * The accessor hands out the list that the provider iterates while it runs. A caller that
+ * mutated the returned list raced that iteration, so the accessor must hand out a copy
+ * instead of the live list.
+ */
+ @Test
+ public void testGetStaticMembersReturnsACopy() throws Exception {
+ StaticMembershipService service = new StaticMembershipService();
+ StaticMember member = new StaticMember("127.0.0.1", 4005, -1);
+ service.addStaticMember(member);
+
+ ArrayList<StaticMember> members = service.getStaticMembers();
+ Assert.assertNotSame("the accessor must not hand out the same list twice", members,
+ service.getStaticMembers());
+ Assert.assertTrue(members.remove(member));
+ Assert.assertFalse(members.contains(member));
+
+ // The live list that the provider reads must be untouched by the copy mutation
+ Assert.assertTrue(service.getStaticMembers().contains(member));
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
(tomcat) 11/20: Warn when static membership properties are replaced while running
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 4f065219314acd728b90cf6fa4d63e1ad982690e
Author: opencode <re...@apache.org>
AuthorDate: Sun Oct 11 12:00:55 2026 +0200
Warn when static membership properties are replaced while running
The membership service exposes its properties as a read write JMX
attribute, as the required setProperties method of the membership
service interface makes the attribute writable. Replacing the
properties while the service ran silently desynchronized the reported
configuration from the live one, for the attribute getters read the
replaced properties while the membership provider keeps the values it
read when it was built at start, and re-reads none of them until it is
rebuilt.
The contract of setProperties is now documented, and replacing the
properties while a provider exists logs a warning naming the
divergence, so an administrative change that does not take effect until
a full stop and start can no longer pass unnoticed.
Add no test, for the only runtime effect is the warning line and the
tribes tests have no log capture to assert it against.
---
.../catalina/tribes/membership/LocalStrings.properties | 1 +
.../catalina/tribes/membership/StaticMembershipService.java | 12 ++++++++++++
2 files changed, 13 insertions(+)
diff --git a/java/org/apache/catalina/tribes/membership/LocalStrings.properties b/java/org/apache/catalina/tribes/membership/LocalStrings.properties
index b7fd3273a8..a5cbad68cb 100644
--- a/java/org/apache/catalina/tribes/membership/LocalStrings.properties
+++ b/java/org/apache/catalina/tribes/membership/LocalStrings.properties
@@ -76,4 +76,5 @@ staticMembershipProvider.stopMembership.sendFailed=Unable to send stop membershi
staticMembershipService.localMemberNotSet=The local member has not been resolved. Either start the channel with a level that includes the receiver or configure the local member explicitly.
staticMembershipService.noLocalMember=There is no localMember in static member list.
+staticMembershipService.propertiesWhileRunning=The properties were replaced while the membership provider is running. The running provider keeps the values it read when it started.
staticMembershipService.stopFail=Unable to stop the static membership service, level: [{0}]
diff --git a/java/org/apache/catalina/tribes/membership/StaticMembershipService.java b/java/org/apache/catalina/tribes/membership/StaticMembershipService.java
index 841a49985a..85fc6b9ae0 100644
--- a/java/org/apache/catalina/tribes/membership/StaticMembershipService.java
+++ b/java/org/apache/catalina/tribes/membership/StaticMembershipService.java
@@ -263,8 +263,20 @@ public class StaticMembershipService extends MembershipServiceBase implements St
properties.setProperty("pingInterval", String.valueOf(pingInterval));
}
+ /**
+ * Sets the properties for this service. The values are read when the membership provider is
+ * built, as the service starts. Replacing the properties while the service runs therefore has
+ * no effect on the running provider. It keeps the values it started with until the service is
+ * fully stopped and started again, even though the attribute getters of the service report the
+ * new values.
+ *
+ * @param properties the new configuration
+ */
@Override
public void setProperties(Properties properties) {
+ if (provider != null) {
+ log.warn(sm.getString("staticMembershipService.propertiesWhileRunning"));
+ }
setDefaults(properties);
this.properties = properties;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
(tomcat) 17/20: Send broadcasts and pings to the checked member snapshot
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 3ff5ca9b1fcbb5a7890de77f202df20001170f0b
Author: opencode <re...@apache.org>
AuthorDate: Sun Oct 11 14:18:31 2026 +0200
Send broadcasts and pings to the checked member snapshot
The broadcast snapshots the channel members and returns early when the
snapshot is empty, but the non RPC branch then sent with a fresh read
of the channel members. A membership change between the check and the
send moved the message to another member set than the checked one, or
lost it entirely when the fresh read came back empty, which silently
dropped the stop notification of the breakdown when the last members
had just left. The ping had the same double read for its RPC send.
The send now uses the members that were checked to be non empty.
---
java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java | 7 ++++---
1 file changed, 4 insertions(+), 3 deletions(-)
diff --git a/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java b/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
index 51dd9981f9..29e0c69147 100644
--- a/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
+++ b/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
@@ -322,10 +322,11 @@ public abstract class AbstractReplicatedMap<K, V>
protected void ping(long timeout) throws ChannelException {
MapMessage msg = new MapMessage(this.mapContextName, MapMessage.MSG_PING, false, null, null, null,
channel.getLocalMember(false), null);
- if (channel.getMembers().length > 0) {
+ Member[] channelMembers = channel.getMembers();
+ if (channelMembers.length > 0) {
try {
// send a ping, wait for all nodes to reply
- Response[] resp = rpcChannel.send(channel.getMembers(), msg, RpcChannel.ALL_REPLY, (channelSendOptions),
+ Response[] resp = rpcChannel.send(channelMembers, msg, RpcChannel.ALL_REPLY, (channelSendOptions),
(int) accessTimeout);
for (Response response : resp) {
MapMessage mapMsg = (MapMessage) response.getMessage();
@@ -429,7 +430,7 @@ public abstract class AbstractReplicatedMap<K, V>
log.warn(sm.getString("abstractReplicatedMap.broadcast.noReplies"));
}
} else {
- channel.send(channel.getMembers(), msg, channelSendOptions);
+ channel.send(members, msg, channelSendOptions);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
(tomcat) 16/20: Scope the map member and state locks of AbstractReplicatedMap to the map operations
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 1d5cd7a559337a92ad6c046c03c2c1d34cbde29d
Author: opencode <re...@apache.org>
AuthorDate: Sun Oct 11 13:36:44 2026 +0200
Scope the map member and state locks of AbstractReplicatedMap to the map operations
The expiry handling of the ping ran the relocation of the entries of a
timed out member while holding the map members lock. That relocation
sends to the remaining members and blocks until they acknowledge, so
one slow or dead member stalled every session operation, since every
put, remove and get reads the map members lock.
Likewise, the backup selection for a newly added member performed its
blocking publications while holding the state lock, and the state
transfer paths held the state lock across the full (de)serialization
of the map content: the deserialization of a received full state, and
the serialization of every serializable entry when answering a state
request for a joining member.
The locks are now held only briefly. The ping collects the expired
members under the lock and then re-checks each one before handling the
disappearance outside of the lock. The member addition collects the
primary entries that have no backup under the state lock and publishes
them outside of it, re-checking that the entry still needs a backup.
The state transfer and the state request only scan the map under the
state lock, and deserialize or serialize outside of it.
Add unit tests that verify that the other thread can enter the monitor
while the expiry, the backup publication and the state serialization
run.
---
.../tribes/tipis/AbstractReplicatedMap.java | 92 ++++++---
.../tribes/tipis/TestAbstractReplicatedMap.java | 214 +++++++++++++++++++++
webapps/docs/changelog.xml | 12 ++
3 files changed, 293 insertions(+), 25 deletions(-)
diff --git a/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java b/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
index 46380ba8b2..51dd9981f9 100644
--- a/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
+++ b/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
@@ -363,17 +363,30 @@ public abstract class AbstractReplicatedMap<K, V>
}
}
// update our map of members, expire some if we didn't receive a ping back
+ List<Member> expiredMembers = new ArrayList<>();
synchronized (mapMembers) {
Member[] members = mapMembers.keySet().toArray(new Member[0]);
long now = System.currentTimeMillis();
for (Member member : members) {
long access = mapMembers.get(member).longValue();
if ((now - access) > timeout) {
- log.warn(sm.getString("abstractReplicatedMap.ping.timeout", member, mapname));
- memberDisappeared(member);
+ expiredMembers.add(member);
}
}
} // synch
+ // The expiry handling relocates entries, which sends to other members and may block
+ // until they acknowledge, so it must not run while holding the map members lock
+ for (Member member : expiredMembers) {
+ boolean expired;
+ synchronized (mapMembers) {
+ Long access = mapMembers.get(member);
+ expired = access != null && (System.currentTimeMillis() - access.longValue()) > timeout;
+ }
+ if (expired) {
+ log.warn(sm.getString("abstractReplicatedMap.ping.timeout", member, mapname));
+ memberDisappeared(member);
+ }
+ }
}
/**
@@ -625,12 +638,26 @@ public abstract class AbstractReplicatedMap<K, V>
Response[] resp = rpcChannel.send(new Member[] { backup }, msg, RpcChannel.FIRST_REPLY,
channelSendOptions, rpcTimeout);
if (resp.length > 0) {
+ msg = (MapMessage) resp[0].getMessage();
+ msg.deserialize(getExternalLoaders());
+ ArrayList<?> list = (ArrayList<?>) msg.getValue();
+ // Deserialize the state messages outside of the state lock below, so that
+ // it is held only for the application of the entries to the map
+ List<Serializable> messages = new ArrayList<>(list.size());
+ for (Object o : list) {
+ if (o instanceof MapMessage mapMessage) {
+ try {
+ mapMessage.deserialize(getExternalLoaders());
+ } catch (ClassNotFoundException | IOException x) {
+ log.error(sm.getString("abstractReplicatedMap.unable.deserialize.MapMessage"), x);
+ continue;
+ }
+ }
+ messages.add((Serializable) o);
+ }
synchronized (stateMutex) {
- msg = (MapMessage) resp[0].getMessage();
- msg.deserialize(getExternalLoaders());
- ArrayList<?> list = (ArrayList<?>) msg.getValue();
- for (Object o : list) {
- messageReceived((Serializable) o, resp[0].getSource());
+ for (Serializable message : messages) {
+ messageReceived(message, resp[0].getSource());
} // for
}
stateTransferred = true;
@@ -683,23 +710,27 @@ public abstract class AbstractReplicatedMap<K, V>
// state transfer request
if (mapmsg.getMsgType() == MapMessage.MSG_STATE || mapmsg.getMsgType() == MapMessage.MSG_STATE_COPY) {
- synchronized (stateMutex) { // make sure we don't do two things at the same time
- ArrayList<MapMessage> list = new ArrayList<>();
+ boolean copy = (mapmsg.getMsgType() == MapMessage.MSG_STATE_COPY);
+ List<MapEntry<K,V>> snapshot = new ArrayList<>();
+ synchronized (stateMutex) { // scan under the lock, so that a concurrent state operation cannot overlap it
for (Entry<K,MapEntry<K,V>> e : innerMap.entrySet()) {
MapEntry<K,V> entry = innerMap.get(e.getKey());
if (entry != null && entry.isSerializable()) {
- boolean copy = (mapmsg.getMsgType() == MapMessage.MSG_STATE_COPY);
- MapMessage me =
- new MapMessage(mapContextName, copy ? MapMessage.MSG_COPY : MapMessage.MSG_PROXY, false,
- (Serializable) entry.getKey(), copy ? (Serializable) entry.getValue() : null,
- null, entry.getPrimary(), entry.getBackupNodes());
- list.add(me);
+ snapshot.add(entry);
}
}
- mapmsg.setValue(list);
- return mapmsg;
-
} // synchronized
+ // Serialize the snapshot outside of the state lock, as serializing the whole
+ // map content blocks every other state operation for a long time otherwise
+ ArrayList<MapMessage> list = new ArrayList<>(snapshot.size());
+ for (MapEntry<K,V> entry : snapshot) {
+ MapMessage me = new MapMessage(mapContextName, copy ? MapMessage.MSG_COPY : MapMessage.MSG_PROXY,
+ false, (Serializable) entry.getKey(), copy ? (Serializable) entry.getValue() : null, null,
+ entry.getPrimary(), entry.getBackupNodes());
+ list.add(me);
+ }
+ mapmsg.setValue(list);
+ return mapmsg;
}
// ping
@@ -930,6 +961,7 @@ public abstract class AbstractReplicatedMap<K, V>
}
}
if (memberAdded) {
+ List<MapEntry<K,V>> entriesWithoutBackup = new ArrayList<>();
synchronized (stateMutex) {
for (Entry<K,MapEntry<K,V>> e : innerMap.entrySet()) {
MapEntry<K,V> entry = innerMap.get(e.getKey());
@@ -937,16 +969,26 @@ public abstract class AbstractReplicatedMap<K, V>
continue;
}
if (entry.isPrimary() && (entry.getBackupNodes() == null || entry.getBackupNodes().length == 0)) {
- try {
- Member[] backup = publishEntryInfo(entry.getKey(), entry.getValue());
- entry.setBackupNodes(backup);
- entry.setPrimary(channel.getLocalMember(false));
- } catch (ChannelException x) {
- log.error(sm.getString("abstractReplicatedMap.unableSelect.backup"), x);
- } // catch
+ entriesWithoutBackup.add(entry);
} // end if
} // while
} // synchronized
+ // Select and publish the backups outside of the state lock, as the replication
+ // blocks until the backup acknowledges, while session operations and the state
+ // transfer of joining members need that lock
+ for (MapEntry<K,V> entry : entriesWithoutBackup) {
+ if (!entry.isPrimary() || (entry.getBackupNodes() != null && entry.getBackupNodes().length > 0)) {
+ // The entry found its backup in the meantime
+ continue;
+ }
+ try {
+ Member[] backup = publishEntryInfo(entry.getKey(), entry.getValue());
+ entry.setBackupNodes(backup);
+ entry.setPrimary(channel.getLocalMember(false));
+ } catch (ChannelException x) {
+ log.error(sm.getString("abstractReplicatedMap.unableSelect.backup"), x);
+ } // catch
+ } // for
} // end if
}
diff --git a/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java b/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java
new file mode 100644
index 0000000000..83d933246b
--- /dev/null
+++ b/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java
@@ -0,0 +1,214 @@
+/*
+ * 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.tipis;
+
+import java.io.IOException;
+import java.io.ObjectOutputStream;
+import java.io.Serial;
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.group.GroupChannel;
+import org.apache.catalina.tribes.membership.MemberImpl;
+import org.apache.catalina.tribes.tipis.AbstractReplicatedMap.MapMessage;
+
+public class TestAbstractReplicatedMap {
+
+ /*
+ * The channel resolves any looked up member directly, so that a member can be added to the map
+ * members without starting a channel.
+ */
+ private static final class TestChannel extends GroupChannel {
+
+ @Override
+ public Member getMember(Member mbr) {
+ return mbr;
+ }
+ }
+
+ /*
+ * A value whose first serialization blocks until it is released, to simulate the long lasting
+ * serialization of a full map content.
+ */
+ private static final class BlockingValue implements Serializable {
+
+ @Serial
+ private static final long serialVersionUID = 1L;
+
+ private final transient CountDownLatch release;
+ private final transient CountDownLatch serializationStarted = new CountDownLatch(1);
+ private transient boolean serializedOnce = false;
+
+ BlockingValue(CountDownLatch release) {
+ this.release = release;
+ }
+
+ private void writeObject(ObjectOutputStream out) throws IOException {
+ out.defaultWriteObject();
+ if (!serializedOnce) {
+ serializedOnce = true;
+ serializationStarted.countDown();
+ try {
+ release.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException(e);
+ }
+ }
+ }
+ }
+
+ /*
+ * Whether another thread could enter the given monitor while the current thread stays inside
+ * the operation that is expected to run outside of that monitor.
+ */
+ private static boolean monitorAvailable(Object monitor, long timeoutMs) {
+ CountDownLatch acquired = new CountDownLatch(1);
+ Thread probe = new Thread(() -> {
+ synchronized (monitor) {
+ acquired.countDown();
+ }
+ });
+ probe.setDaemon(true);
+ probe.start();
+ try {
+ return acquired.await(timeoutMs, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
+
+ /*
+ * The expiry handling of the ping aborts the entries of a timed out member, which sends to the
+ * remaining members and may block until they acknowledge. It must not run while holding the
+ * map members lock, since every session operation needs that lock.
+ */
+ @Test
+ public void testPingExpiresOutsideMapMembersLock() throws Exception {
+ Member member = new MemberImpl("localhost", 4011, -1);
+ AtomicInteger disappeared = new AtomicInteger();
+ AtomicBoolean lockAvailable = new AtomicBoolean();
+
+ LazyReplicatedMap<String,String> map = new LazyReplicatedMap<String,String>(
+ (key, value) -> {
+ // Not made primary here
+ }, new TestChannel(), 1000, "TestAbstractReplicatedMap", null) {
+
+ @Override
+ public void memberDisappeared(Member disappearedMember) {
+ disappeared.incrementAndGet();
+ lockAvailable.set(monitorAvailable(mapMembers, 2000));
+ }
+ };
+
+ map.memberAlive(member);
+ Thread.sleep(10);
+ // A timeout of zero expires every member that was not refreshed just now
+ map.ping(0);
+
+ Assert.assertEquals(1, disappeared.get());
+ Assert.assertTrue("The ping must expire members outside the map members lock", lockAvailable.get());
+ }
+
+ /*
+ * Adding a member selects backups for the primary entries that have none, which blocks until
+ * the selected backups acknowledge. That must not run while holding the state lock, since the
+ * state transfer of joining members needs that lock.
+ */
+ @Test
+ public void testMapMemberAddedPublishesOutsideStateLock() throws Exception {
+ AtomicBoolean probeOnPublish = new AtomicBoolean();
+ AtomicBoolean lockAvailable = new AtomicBoolean();
+ AtomicInteger published = new AtomicInteger();
+
+ LazyReplicatedMap<String,String> map = new LazyReplicatedMap<String,String>(
+ (key, value) -> {
+ // Not made primary here
+ }, new TestChannel(), 1000, "TestAbstractReplicatedMap", null) {
+
+ @Override
+ protected Member[] publishEntryInfo(Object key, Object value) {
+ if (probeOnPublish.get()) {
+ lockAvailable.set(monitorAvailable(stateMutex, 2000));
+ }
+ published.incrementAndGet();
+ return new Member[0];
+ }
+ };
+
+ // Stores a primary entry without a backup, without probing the publication it attempts
+ map.put("key", "value");
+ probeOnPublish.set(true);
+ map.mapMemberAdded(new MemberImpl("localhost", 4012, -1));
+
+ Assert.assertEquals(2, published.get());
+ Assert.assertTrue("The member addition must publish backups outside the state lock", lockAvailable.get());
+ }
+
+ /*
+ * Answering a state request serializes the whole map content, which must not happen while
+ * holding the state lock, since it blocks the state transfer of other joining members for the
+ * duration of that serialization.
+ */
+ @Test
+ public void testStateRequestSerializesOutsideStateLock() throws Exception {
+ CountDownLatch release = new CountDownLatch(1);
+ BlockingValue value = new BlockingValue(release);
+
+ LazyReplicatedMap<String,BlockingValue> map = new LazyReplicatedMap<String,BlockingValue>(
+ (key, val) -> {
+ // Not made primary here
+ }, new TestChannel(), 1000, "TestAbstractReplicatedMap", null);
+ map.put("key", value);
+
+ MapMessage stateRequest = new MapMessage("TestAbstractReplicatedMap".getBytes(StandardCharsets.UTF_8),
+ MapMessage.MSG_STATE_COPY, false, null, null, null, null, null);
+ AtomicReference<Serializable> reply = new AtomicReference<>();
+ Member sender = new MemberImpl("localhost", 4013, -1);
+ Thread replyThread = new Thread(() -> reply.set(map.replyRequest(stateRequest, sender)));
+ replyThread.setDaemon(true);
+ replyThread.start();
+
+ // Once the value serialization has started, the map scan is complete
+ try {
+ Assert.assertTrue("The state request must serialize the map content",
+ value.serializationStarted.await(5, TimeUnit.SECONDS));
+ boolean available = monitorAvailable(map.stateMutex, 2000);
+ Assert.assertTrue("The state request must serialize outside the state lock", available);
+ } finally {
+ release.countDown();
+ }
+ replyThread.join(5000);
+ Assert.assertTrue(reply.get() instanceof MapMessage);
+ Object list = ((MapMessage) reply.get()).getValue();
+ Assert.assertTrue(list instanceof ArrayList);
+ Assert.assertEquals(1, ((ArrayList<?>) list).size());
+ Assert.assertTrue(((ArrayList<?>) list).get(0) instanceof MapMessage);
+ Assert.assertEquals(MapMessage.MSG_COPY, ((MapMessage) ((ArrayList<?>) list).get(0)).getMsgType());
+ }
+}
diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml
index fd2201a78e..dd6e9f20fd 100644
--- a/webapps/docs/changelog.xml
+++ b/webapps/docs/changelog.xml
@@ -2593,6 +2593,18 @@
further confirmation ever followed, so a single moment of
network reordering cost one lost cluster message. (remm)
</fix>
+ <fix>
+ The <code>AbstractReplicatedMap</code> now holds the map
+ members lock and the state lock only briefly. The entry
+ relocation triggered by the ping expiry runs outside the map
+ members lock, the backup selection for a newly added member
+ outside the state lock, and the (de)serialization of the map
+ content for a member joining or transferring its state
+ likewise outside the state lock. Previously, a slow send to a
+ dead member or the serialization of a large map blocked every
+ session operation and the state transfer of joining members
+ for the duration of that work. (remm)
+ </fix>
</changelog>
</subsection>
<subsection name="Other">
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
(tomcat) 03/20: Avoid NullPointerException when stopping channel components without a registry
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit f74d9985e6f7c8eca9c1da3d14b777bd5ef7e4d3
Author: opencode <re...@apache.org>
AuthorDate: Sun Oct 11 10:21:38 2026 +0200
Avoid NullPointerException when stopping channel components without a registry
The channel components guarded their JMX deregistration at stop with
their own ObjectName field, a hint set at start time, and then called
JmxRegistry.getRegistry() unconditionally. The registry, however, is
removed from the cache as soon as the channel MBean is deregistered
externally, and getRegistry() then returns null once the JMX
configuration no longer allows a new registry to be created, for
example after the channel was renamed or JMX was disabled. The
resulting NullPointerException aborted the stop of the channel, the
interceptors, the transmitter and the receiver, or was swallowed by
the membership services whose remaining cleanup, releasing the
provider and the channel reference, was then skipped.
The registry is now null-checked at all seven stop paths and only the
deregistration itself is skipped when it is absent, as it is the only
vehicle for unregistering. The rest of the stop always runs and the
stale ObjectName is cleared.
Add tests for the recreate-and-unregister path and for the invalidated
configuration, the latter reproducing the NullPointerException on the
previous code.
---
.../tribes/group/ChannelInterceptorBase.java | 5 +-
.../apache/catalina/tribes/group/GroupChannel.java | 5 +-
.../catalina/tribes/membership/McastService.java | 5 +-
.../tribes/membership/StaticMembershipService.java | 5 +-
.../membership/cloud/CloudMembershipService.java | 5 +-
.../catalina/tribes/transport/ReceiverBase.java | 5 +-
.../tribes/transport/ReplicationTransmitter.java | 5 +-
.../catalina/tribes/jmx/TestJmxRegistry.java | 103 +++++++++++++++++++++
8 files changed, 131 insertions(+), 7 deletions(-)
diff --git a/java/org/apache/catalina/tribes/group/ChannelInterceptorBase.java b/java/org/apache/catalina/tribes/group/ChannelInterceptorBase.java
index 3e34f8f568..c535233675 100644
--- a/java/org/apache/catalina/tribes/group/ChannelInterceptorBase.java
+++ b/java/org/apache/catalina/tribes/group/ChannelInterceptorBase.java
@@ -188,7 +188,10 @@ public abstract class ChannelInterceptorBase implements ChannelInterceptor {
getNext().stop(svc);
}
if (oname != null) {
- JmxRegistry.getRegistry(channel).unregisterJmx(oname);
+ JmxRegistry jmxRegistry = JmxRegistry.getRegistry(channel);
+ if (jmxRegistry != null) {
+ jmxRegistry.unregisterJmx(oname);
+ }
oname = null;
}
channel = null;
diff --git a/java/org/apache/catalina/tribes/group/GroupChannel.java b/java/org/apache/catalina/tribes/group/GroupChannel.java
index b790ccf7c4..99eb3be4c9 100644
--- a/java/org/apache/catalina/tribes/group/GroupChannel.java
+++ b/java/org/apache/catalina/tribes/group/GroupChannel.java
@@ -503,7 +503,10 @@ public class GroupChannel extends ChannelInterceptorBase implements ManagedChann
ownExecutor = false;
}
if (oname != null) {
- JmxRegistry.getRegistry(this).unregisterJmx(oname);
+ JmxRegistry jmxRegistry = JmxRegistry.getRegistry(this);
+ if (jmxRegistry != null) {
+ jmxRegistry.unregisterJmx(oname);
+ }
oname = null;
}
}
diff --git a/java/org/apache/catalina/tribes/membership/McastService.java b/java/org/apache/catalina/tribes/membership/McastService.java
index a368eae81f..cda9f12608 100644
--- a/java/org/apache/catalina/tribes/membership/McastService.java
+++ b/java/org/apache/catalina/tribes/membership/McastService.java
@@ -418,7 +418,10 @@ public class McastService extends MembershipServiceBase implements MessageListen
try {
if (impl != null && impl.stop(svc)) {
if (oname != null) {
- JmxRegistry.getRegistry(channel).unregisterJmx(oname);
+ JmxRegistry jmxRegistry = JmxRegistry.getRegistry(channel);
+ if (jmxRegistry != null) {
+ jmxRegistry.unregisterJmx(oname);
+ }
oname = null;
}
impl.setChannel(null);
diff --git a/java/org/apache/catalina/tribes/membership/StaticMembershipService.java b/java/org/apache/catalina/tribes/membership/StaticMembershipService.java
index 2eb1df86bf..74c0405eb0 100644
--- a/java/org/apache/catalina/tribes/membership/StaticMembershipService.java
+++ b/java/org/apache/catalina/tribes/membership/StaticMembershipService.java
@@ -102,7 +102,10 @@ public class StaticMembershipService extends MembershipServiceBase implements St
try {
if (provider != null && provider.stop(level)) {
if (oname != null) {
- JmxRegistry.getRegistry(channel).unregisterJmx(oname);
+ JmxRegistry jmxRegistry = JmxRegistry.getRegistry(channel);
+ if (jmxRegistry != null) {
+ jmxRegistry.unregisterJmx(oname);
+ }
oname = null;
}
provider = null;
diff --git a/java/org/apache/catalina/tribes/membership/cloud/CloudMembershipService.java b/java/org/apache/catalina/tribes/membership/cloud/CloudMembershipService.java
index 737503dd50..15d0fd45b0 100644
--- a/java/org/apache/catalina/tribes/membership/cloud/CloudMembershipService.java
+++ b/java/org/apache/catalina/tribes/membership/cloud/CloudMembershipService.java
@@ -180,7 +180,10 @@ public class CloudMembershipService extends MembershipServiceBase implements Clo
try {
if (membershipProvider != null && membershipProvider.stop(level)) {
if (oname != null) {
- JmxRegistry.getRegistry(channel).unregisterJmx(oname);
+ JmxRegistry jmxRegistry = JmxRegistry.getRegistry(channel);
+ if (jmxRegistry != null) {
+ jmxRegistry.unregisterJmx(oname);
+ }
oname = null;
}
membershipProvider = null;
diff --git a/java/org/apache/catalina/tribes/transport/ReceiverBase.java b/java/org/apache/catalina/tribes/transport/ReceiverBase.java
index 85166eb824..f3f21ddbef 100644
--- a/java/org/apache/catalina/tribes/transport/ReceiverBase.java
+++ b/java/org/apache/catalina/tribes/transport/ReceiverBase.java
@@ -139,7 +139,10 @@ public abstract class ReceiverBase implements ChannelReceiver, ListenCallback, R
}
executor = null;
if (oname != null) {
- JmxRegistry.getRegistry(channel).unregisterJmx(oname);
+ JmxRegistry jmxRegistry = JmxRegistry.getRegistry(channel);
+ if (jmxRegistry != null) {
+ jmxRegistry.unregisterJmx(oname);
+ }
oname = null;
}
channel = null;
diff --git a/java/org/apache/catalina/tribes/transport/ReplicationTransmitter.java b/java/org/apache/catalina/tribes/transport/ReplicationTransmitter.java
index dfc9b5a83a..64a729ef07 100644
--- a/java/org/apache/catalina/tribes/transport/ReplicationTransmitter.java
+++ b/java/org/apache/catalina/tribes/transport/ReplicationTransmitter.java
@@ -99,7 +99,10 @@ public class ReplicationTransmitter implements ChannelSender {
public synchronized void stop() {
getTransport().disconnect();
if (oname != null) {
- JmxRegistry.getRegistry(channel).unregisterJmx(oname);
+ JmxRegistry jmxRegistry = JmxRegistry.getRegistry(channel);
+ if (jmxRegistry != null) {
+ jmxRegistry.unregisterJmx(oname);
+ }
oname = null;
}
channel = null;
diff --git a/test/org/apache/catalina/tribes/jmx/TestJmxRegistry.java b/test/org/apache/catalina/tribes/jmx/TestJmxRegistry.java
new file mode 100644
index 0000000000..fa44f85e45
--- /dev/null
+++ b/test/org/apache/catalina/tribes/jmx/TestJmxRegistry.java
@@ -0,0 +1,103 @@
+/*
+ * 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.jmx;
+
+import java.lang.management.ManagementFactory;
+
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import org.apache.catalina.tribes.Channel;
+import org.apache.catalina.tribes.group.ChannelInterceptorBase;
+import org.apache.catalina.tribes.group.GroupChannel;
+
+public class TestJmxRegistry {
+
+ public interface TestInterceptorMBean {
+ // the presence of the companion interface is what makes the interceptor
+ // registerable as a Standard MBean, as for the production interceptors
+ }
+
+ public static class TestInterceptor extends ChannelInterceptorBase implements TestInterceptorMBean {
+ }
+
+ /*
+ * A component ObjectName is a stale hint: the registry can be gone when the component stops,
+ * because it is removed when the channel MBean is deregistered externally. If the JMX
+ * configuration is still valid, the registry is simply recreated and the MBean unregistered.
+ */
+ @Test
+ public void testStopRecreatesRegistryWhenStillConfigured() throws Exception {
+ GroupChannel channel = new GroupChannel();
+ channel.setName("TestJmxRegistry-" + System.nanoTime());
+ MBeanServer mbserver = ManagementFactory.getPlatformMBeanServer();
+ try {
+ TestInterceptor interceptor = new TestInterceptor();
+ interceptor.setChannel(channel);
+ interceptor.start(Channel.DEFAULT);
+ ObjectName bean = new ObjectName("ClusterChannel:type=Channel,channel=" + channel.getName() +
+ ",component=Interceptor,interceptorName=TestInterceptor");
+ Assert.assertTrue(mbserver.isRegistered(bean));
+
+ JmxRegistry.removeRegistry(channel, false);
+
+ interceptor.stop(Channel.DEFAULT);
+ Assert.assertNull(interceptor.getChannel());
+ Assert.assertFalse(mbserver.isRegistered(bean));
+ } finally {
+ JmxRegistry.removeRegistry(channel, true);
+ }
+ }
+
+ /*
+ * If the JMX configuration is invalidated after the registry was removed, the components
+ * still have to complete their stop and their cleanup. Previously, the missing registry
+ * surfaced as an uncaught NullPointerException.
+ */
+ @Test
+ public void testStopWithoutRegistryDoesNotThrow() throws Exception {
+ GroupChannel channel = new GroupChannel();
+ channel.setName("TestJmxRegistry-" + System.nanoTime());
+ MBeanServer mbserver = ManagementFactory.getPlatformMBeanServer();
+ ObjectName bean = new ObjectName("ClusterChannel:type=Channel,channel=" + channel.getName() +
+ ",component=Interceptor,interceptorName=TestInterceptor");
+ try {
+ TestInterceptor interceptor = new TestInterceptor();
+ interceptor.setChannel(channel);
+ interceptor.start(Channel.DEFAULT);
+ Assert.assertTrue(mbserver.isRegistered(bean));
+
+ // Simulate the channel MBean being deregistered externally, which removes the
+ // registry, followed by an invalidation of the JMX configuration
+ JmxRegistry.removeRegistry(channel, false);
+ channel.setJmxEnabled(false);
+
+ interceptor.stop(Channel.DEFAULT);
+ Assert.assertNull(interceptor.getChannel());
+ } finally {
+ // Without a registry the stop path cannot unregister the bean any more, clean it up
+ // here so the shared MBean server is not polluted
+ if (mbserver.isRegistered(bean)) {
+ mbserver.unregisterMBean(bean);
+ }
+ JmxRegistry.removeRegistry(channel, true);
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
(tomcat) 18/20: Skip the removal notification for keys that cannot be serialized
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 523945e31db9a8d84a53da11b2dd68068b281303
Author: opencode <re...@apache.org>
AuthorDate: Sun Oct 11 14:26:19 2026 +0200
Skip the removal notification for keys that cannot be serialized
The removal broadcast built its message with the key, which threw a
class cast exception for a key that does not implement Serializable, or
a runtime exception when a Serializable key still failed to serialize.
The exception was not caught as a channel exception and escaped the
removal. Since the put calls the removal first, a put that replaced an
existing entry under such a key aborted before it stored the new value.
Keys that cannot be serialized only ever exist on the node that stored
them, as every replication path checks that the key is serializable
before it sends one, so the notification has no destination that could
hold the entry. The removal now skips it, matching the put which also
silently keeps such an entry local.
---
.../tribes/tipis/AbstractReplicatedMap.java | 4 ++-
.../tribes/tipis/TestAbstractReplicatedMap.java | 33 ++++++++++++++++++++++
2 files changed, 36 insertions(+), 1 deletion(-)
diff --git a/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java b/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
index 29e0c69147..ead1263dd8 100644
--- a/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
+++ b/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
@@ -1222,7 +1222,9 @@ public abstract class AbstractReplicatedMap<K, V>
MapEntry<K,V> entry = innerMap.remove(key);
try {
- if (getMapMembers().length > 0 && notify && entry != null) {
+ // Skip the notification for a key that cannot be serialized, as the message
+ // construction would fail on it, like the replication of put does
+ if (getMapMembers().length > 0 && notify && entry != null && key instanceof Serializable) {
MapMessage msg = new MapMessage(getMapContextName(), MapMessage.MSG_REMOVE, false, (Serializable) key,
null, null, null, null);
getChannel().send(getMapMembers(), msg, getChannelSendOptions());
diff --git a/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java b/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java
index 83d933246b..e4da5f26fa 100644
--- a/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java
+++ b/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java
@@ -31,7 +31,9 @@ import java.util.concurrent.atomic.AtomicReference;
import org.junit.Assert;
import org.junit.Test;
+import org.apache.catalina.tribes.ErrorHandler;
import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.UniqueId;
import org.apache.catalina.tribes.group.GroupChannel;
import org.apache.catalina.tribes.membership.MemberImpl;
import org.apache.catalina.tribes.tipis.AbstractReplicatedMap.MapMessage;
@@ -44,10 +46,18 @@ public class TestAbstractReplicatedMap {
*/
private static final class TestChannel extends GroupChannel {
+ private final AtomicInteger sendCalls = new AtomicInteger();
+
@Override
public Member getMember(Member mbr) {
return mbr;
}
+
+ @Override
+ public UniqueId send(Member[] destination, Serializable msg, int options, ErrorHandler handler) {
+ sendCalls.incrementAndGet();
+ return new UniqueId();
+ }
}
/*
@@ -211,4 +221,27 @@ public class TestAbstractReplicatedMap {
Assert.assertTrue(((ArrayList<?>) list).get(0) instanceof MapMessage);
Assert.assertEquals(MapMessage.MSG_COPY, ((MapMessage) ((ArrayList<?>) list).get(0)).getMsgType());
}
+
+ /*
+ * The put silently accepts keys that cannot be replicated, so the remove must not fail on such
+ * a key either, but just skip the notification to the other members. The removal previously
+ * threw an unchecked exception from the message construction, which also escaped a put that
+ * replaced such an entry before the new value was stored.
+ */
+ @Test
+ public void testRemoveNonSerializableKeySkipsNotification() throws Exception {
+ TestChannel channel = new TestChannel();
+ LazyReplicatedMap<Object,String> map = new LazyReplicatedMap<Object,String>(
+ (key, value) -> {
+ // Not made primary here
+ }, channel, 1000, "TestAbstractReplicatedMap", null);
+ map.memberAlive(new MemberImpl("localhost", 4016, -1));
+
+ Object key = new Object();
+ map.put(key, "value");
+
+ Assert.assertEquals("value", map.remove(key));
+ Assert.assertNull(map.getInternal(key));
+ Assert.assertEquals(0, channel.sendCalls.get());
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
(tomcat) 10/20: Fail clearly when static membership starts without a local member
Posted by re...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 1e8fea97f9de3c425ac2423bf2c6e5bf9f57971a
Author: opencode <re...@apache.org>
AuthorDate: Sun Oct 11 11:41:40 2026 +0200
Fail clearly when static membership starts without a local member
The static membership service dereferenced its local member when it
started, although the member is only known once the channel receiver
has started, or when it was configured explicitly. Starting the
membership at a level that does not include the receiver therefore
ended in a null pointer exception wrapped into a channel exception,
with no hint at the missing configuration.
The service now reports an illegal state exception naming the two ways
of providing the local member, before building anything, so the failed
start leaves the service untouched and a later start that resolves the
member works as before.
Add a test that a start with an unknown local member fails with that
exception and leaves no provider behind.
---
.../tribes/membership/LocalStrings.properties | 1 +
.../tribes/membership/StaticMembershipService.java | 5 +++++
.../membership/TestStaticMembershipService.java | 21 +++++++++++++++++++++
3 files changed, 27 insertions(+)
diff --git a/java/org/apache/catalina/tribes/membership/LocalStrings.properties b/java/org/apache/catalina/tribes/membership/LocalStrings.properties
index 97d1dbdb5b..b7fd3273a8 100644
--- a/java/org/apache/catalina/tribes/membership/LocalStrings.properties
+++ b/java/org/apache/catalina/tribes/membership/LocalStrings.properties
@@ -74,5 +74,6 @@ staticMembershipProvider.replyRequest.ignored=Message [{0}] is ignored.
staticMembershipProvider.startMembership.noReplies=Received 0 replies, probably a timeout.
staticMembershipProvider.stopMembership.sendFailed=Unable to send stop membership message.
+staticMembershipService.localMemberNotSet=The local member has not been resolved. Either start the channel with a level that includes the receiver or configure the local member explicitly.
staticMembershipService.noLocalMember=There is no localMember in static member list.
staticMembershipService.stopFail=Unable to stop the static membership service, level: [{0}]
diff --git a/java/org/apache/catalina/tribes/membership/StaticMembershipService.java b/java/org/apache/catalina/tribes/membership/StaticMembershipService.java
index 74c0405eb0..841a49985a 100644
--- a/java/org/apache/catalina/tribes/membership/StaticMembershipService.java
+++ b/java/org/apache/catalina/tribes/membership/StaticMembershipService.java
@@ -68,6 +68,11 @@ public class StaticMembershipService extends MembershipServiceBase implements St
provider.start(level);
return;
}
+ if (localMember == null) {
+ // The local member is resolved when the channel receiver starts, or is configured
+ // explicitly. Starting the membership without either leaves it unknown.
+ throw new IllegalStateException(sm.getString("staticMembershipService.localMemberNotSet"));
+ }
localMember.setServiceStartTime(System.currentTimeMillis());
localMember.setMemberAliveTime(100);
// build membership provider
diff --git a/test/org/apache/catalina/tribes/membership/TestStaticMembershipService.java b/test/org/apache/catalina/tribes/membership/TestStaticMembershipService.java
index c2b3992f18..a72fc0b7c7 100644
--- a/test/org/apache/catalina/tribes/membership/TestStaticMembershipService.java
+++ b/test/org/apache/catalina/tribes/membership/TestStaticMembershipService.java
@@ -74,4 +74,25 @@ public class TestStaticMembershipService {
service.start(level);
service.stop(level);
}
+
+ /*
+ * The local member is resolved when the channel receiver starts, or is configured
+ * explicitly. Starting the membership alone, without either, must fail with a clear error
+ * rather than the null pointer exception that dereferencing the missing member caused.
+ */
+ @Test
+ public void testStartWithoutLocalMemberFails() throws Exception {
+ GroupChannel channel = new GroupChannel();
+ channel.setName("TestStaticMembershipService-" + System.nanoTime());
+
+ StaticMembershipService service = new StaticMembershipService();
+ service.setChannel(channel);
+ try {
+ service.start(Channel.MBR_TX_SEQ);
+ Assert.fail("start() must fail when the local member is unknown");
+ } catch (IllegalStateException expected) {
+ // A clear configuration error rather than a null pointer exception
+ }
+ Assert.assertNull(service.getMembershipProvider());
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org