From c6d7d070c59d81db8949683d3e5670b909efb48c Mon Sep 17 00:00:00 2001 From: Doug Rohrer Date: Wed, 23 Aug 2023 18:03:30 -0600 Subject: [PATCH] IsolatedJMX should not release all TCPEndpoints on instance shutdown patch by Doug Rohrer; reviewed by Jon Meredith, Stefan Miklosovic for CASSANDRA-18725 --- build.xml | 2 +- .../utils/RMIClientSocketFactoryImpl.java | 23 +- .../cassandra/utils/ReflectionUtils.java | 101 ++++++++ .../cassandra/distributed/impl/Instance.java | 5 +- .../distributed/impl/IsolatedJmx.java | 95 ++++---- .../distributed/shared/ClusterUtils.java | 226 ++++++++++++++++++ .../distributed/shared/WithProperties.java | 112 +++++++++ .../distributed/test/ResourceLeakTest.java | 54 +++-- .../distributed/test/jmx/JMXFeatureTest.java | 57 ++++- .../test/jmx/JMXGetterCheckTest.java | 48 +++- 10 files changed, 627 insertions(+), 96 deletions(-) create mode 100644 src/java/org/apache/cassandra/utils/ReflectionUtils.java create mode 100644 test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java create mode 100644 test/distributed/org/apache/cassandra/distributed/shared/WithProperties.java diff --git a/build.xml b/build.xml index c634761566..70a195f98d 100644 --- a/build.xml +++ b/build.xml @@ -383,7 +383,7 @@ - + diff --git a/src/java/org/apache/cassandra/utils/RMIClientSocketFactoryImpl.java b/src/java/org/apache/cassandra/utils/RMIClientSocketFactoryImpl.java index 62ab88fb92..c04e934fdc 100644 --- a/src/java/org/apache/cassandra/utils/RMIClientSocketFactoryImpl.java +++ b/src/java/org/apache/cassandra/utils/RMIClientSocketFactoryImpl.java @@ -21,8 +21,11 @@ package org.apache.cassandra.utils; import java.io.IOException; import java.io.Serializable; import java.net.InetAddress; +import java.net.ServerSocket; import java.net.Socket; import java.rmi.server.RMIClientSocketFactory; +import java.util.ArrayList; +import java.util.List; import java.util.Objects; /** @@ -32,6 +35,7 @@ import java.util.Objects; */ public class RMIClientSocketFactoryImpl implements RMIClientSocketFactory, Serializable { + List sockets = new ArrayList<>(); private final InetAddress localAddress; public RMIClientSocketFactoryImpl(InetAddress localAddress) @@ -42,7 +46,24 @@ public class RMIClientSocketFactoryImpl implements RMIClientSocketFactory, Seria @Override public Socket createSocket(String host, int port) throws IOException { - return new Socket(localAddress, port); + Socket socket = new Socket(localAddress, port); + sockets.add(socket); + return socket; + } + + public void close() throws IOException + { + for (Socket socket: sockets) + { + try + { + socket.close(); + } + catch (IOException ignored) + { + // intentionally ignored + } + } } @Override diff --git a/src/java/org/apache/cassandra/utils/ReflectionUtils.java b/src/java/org/apache/cassandra/utils/ReflectionUtils.java new file mode 100644 index 0000000000..3d1cc29565 --- /dev/null +++ b/src/java/org/apache/cassandra/utils/ReflectionUtils.java @@ -0,0 +1,101 @@ +/* + * 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.cassandra.utils; + +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.util.Iterator; +import java.util.Map; +import java.util.function.Predicate; + +public class ReflectionUtils +{ + public static Field getField(Class clazz, String fieldName) throws NoSuchFieldException + { + // below code works before Java 12 + try + { + return clazz.getDeclaredField(fieldName); + } + catch (NoSuchFieldException e) + { + // this is mitigation for JDK 17 (https://bugs.openjdk.org/browse/JDK-8210522) + try + { + Method getDeclaredFields0 = Class.class.getDeclaredMethod("getDeclaredFields0", boolean.class); + getDeclaredFields0.setAccessible(true); + Field[] fields = (Field[]) getDeclaredFields0.invoke(clazz, false); + for (Field field : fields) + { + if (fieldName.equals(field.getName())) + { + return field; + } + } + } + catch (ReflectiveOperationException ex) + { + e.addSuppressed(ex); + } + throw e; + } + } + + /** + * Used by the in-jvm dtest framework to remove entries from private map fields that otherwise would prevent + * collection of classloaders (which causes metaspace OOMs) or otherwise interfere with instance restart. + * + * @param clazz The class which has the map field to clear + * @param instance an instance of the class to clear (pass null for a static member) + * @param mapName the name of the map field to clear + * @param shouldRemove a predicate which determines if the entry in question should be removed + * @param The type of the map key + * @param The type of the map value + */ + public static void clearMapField(Class clazz, Object instance, String mapName, Predicate> shouldRemove) + { + try + { + Field mapField = getField(clazz, mapName); + mapField.setAccessible(true); + // noinspection unchecked + Map map = (Map) mapField.get(instance); + // Because multiple instances can be shutting down at once, + // synchronize on the map to avoid ConcurrentModificationException + synchronized (map) + { + // This could be done with a simple `map.entrySet.removeIf()` call + // but for debugging purposes it is much easier to keep it like this. + Iterator> it = map.entrySet().iterator(); + while (it.hasNext()) + { + Map.Entry entry = it.next(); + if (shouldRemove.test(entry)) + { + it.remove(); + } + } + } + } + catch (NoSuchFieldException | IllegalAccessException ex) + { + throw new RuntimeException(String.format("Could not clear map field %s in class %s", mapName, clazz), ex); + } + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java index cca6296629..bbf98ca022 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java @@ -614,17 +614,18 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance initialized = true; } - private void startJmx() + private synchronized void startJmx() { isolatedJmx = new IsolatedJmx(this, inInstancelogger); isolatedJmx.startJmx(); } - private void stopJmx() throws IllegalAccessException, NoSuchFieldException, InterruptedException + private synchronized void stopJmx() { if (config.has(JMX)) { isolatedJmx.stopJmx(); + isolatedJmx = null; } } diff --git a/test/distributed/org/apache/cassandra/distributed/impl/IsolatedJmx.java b/test/distributed/org/apache/cassandra/distributed/impl/IsolatedJmx.java index b3d06590a9..e586cae58d 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/IsolatedJmx.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/IsolatedJmx.java @@ -18,14 +18,13 @@ package org.apache.cassandra.distributed.impl; -import java.lang.reflect.Field; +import java.io.IOException; import java.net.InetAddress; -import java.net.MalformedURLException; import java.util.HashMap; -import java.util.Iterator; +import java.util.LinkedList; import java.util.Map; +import java.util.concurrent.TimeUnit; import javax.management.remote.JMXConnector; -import javax.management.remote.JMXConnectorFactory; import javax.management.remote.JMXConnectorServer; import javax.management.remote.JMXServiceURL; import javax.management.remote.rmi.RMIConnectorServer; @@ -34,6 +33,8 @@ import javax.management.remote.rmi.RMIJRMPServerImpl; import org.slf4j.Logger; import org.apache.cassandra.distributed.api.IInstanceConfig; +import org.apache.cassandra.distributed.shared.JMXUtil; +import org.apache.cassandra.distributed.shared.Uninterruptibles; import org.apache.cassandra.utils.JMXServerUtils; import org.apache.cassandra.utils.MBeanWrapper; import org.apache.cassandra.utils.RMIClientSocketFactoryImpl; @@ -41,6 +42,7 @@ import sun.rmi.transport.tcp.TCPEndpoint; import static org.apache.cassandra.distributed.api.Feature.JMX; import static org.apache.cassandra.utils.MBeanWrapper.IS_DISABLED_MBEAN_REGISTRATION; +import static org.apache.cassandra.utils.ReflectionUtils.clearMapField; public class IsolatedJmx { @@ -48,7 +50,8 @@ public class IsolatedJmx private static final String SUN_RMI_TRANSPORT_TCP_THREADKEEPALIVETIME = "sun.rmi.transport.tcp.threadKeepAliveTime"; /** Controls the distributed garbage collector lease time for JMX objects. */ private static final String JAVA_RMI_DGC_LEASE_VALUE_IN_JVM_DTEST ="java.rmi.dgc.leaseValue"; - private static final int RMI_KEEPALIVE_TIME = 1000; + public static final int RMI_KEEPALIVE_TIME = 1000; + public static final String UNKNOWN_JMX_CONNECTION_ERROR = "Could not connect to JMX due to an unknown error"; private JMXConnectorServer jmxConnectorServer; private JMXServerUtils.JmxRegistry registry; @@ -59,7 +62,8 @@ public class IsolatedJmx private Logger inInstancelogger; private IInstanceConfig config; - public IsolatedJmx(Instance instance, Logger inInstancelogger) { + public IsolatedJmx(Instance instance, Logger inInstancelogger) + { this.inInstancelogger = inInstancelogger; config = instance.config(); } @@ -127,45 +131,29 @@ public class IsolatedJmx registry.setRemoteServerStub(jmxRmiServer.toStub()); JMXServerUtils.logJmxServiceUrl(addr, jmxPort); - waitForJmxAvailability(hostname, jmxPort, env); + waitForJmxAvailability(env); } - catch (Throwable e) + catch (Throwable t) { - throw new RuntimeException("Feature.JMX was enabled but could not be started.", e); + throw new RuntimeException("Feature.JMX was enabled but could not be started.", t); } } - - private void waitForJmxAvailability(String hostname, int rmiPort, Map env) throws InterruptedException, MalformedURLException + private void waitForJmxAvailability(Map env) { - String url = String.format("service:jmx:rmi:///jndi/rmi://%s:%d/jmxrmi", hostname, rmiPort); - JMXServiceURL serviceURL = new JMXServiceURL(url); - int attempts = 0; - Throwable lastThrown = null; - while (attempts < 20) + try (JMXConnector ignored = JMXUtil.getJmxConnector(config, 20, env)) { - attempts++; - try (JMXConnector ignored = JMXConnectorFactory.connect(serviceURL, env)) - { - inInstancelogger.info("Connected to JMX server at {} after {} attempt(s)", - url, attempts); - return; - } - catch (MalformedURLException e) - { - throw new RuntimeException(e); - } - catch (Throwable thrown) - { - lastThrown = thrown; - } - inInstancelogger.info("Could not connect to JMX on {} after {} attempts. Will retry.", url, attempts); - Thread.sleep(1000); + // Do nothing - JMXUtil now retries + } + catch (IOException iex) + { + // If we land here, there's something more than a timeout + inInstancelogger.error(UNKNOWN_JMX_CONNECTION_ERROR, iex); + throw new RuntimeException(UNKNOWN_JMX_CONNECTION_ERROR, iex); } - throw new RuntimeException("Could not start JMX - unreachable after 20 attempts", lastThrown); } - public void stopJmx() throws IllegalAccessException, NoSuchFieldException, InterruptedException + public void stopJmx() { if (!config.has(JMX)) return; @@ -198,6 +186,14 @@ public class IsolatedJmx inInstancelogger.warn("failed to close registry.", e); } try + { + clientSocketFactory.close(); + } + catch (Throwable e) + { + inInstancelogger.warn("failed to close clientSocketFactory.", e); + } + try { serverSocketFactory.close(); } @@ -207,25 +203,18 @@ public class IsolatedJmx } // The TCPEndpoint class holds references to a class in the in-jvm dtest framework // which transitively has a reference to the InstanceClassLoader, so we need to - // make sure to remove the reference to them when the instance is shutting down - clearMapField(TCPEndpoint.class, null, "localEndpoints"); - Thread.sleep(2 * RMI_KEEPALIVE_TIME); // Double the keep-alive time to give Distributed GC some time to clean up + // make sure to remove the reference to them when the instance is shutting down. + // Additionally, we must make sure to only clear endpoints created by this instance + // As clearning the entire map can cause issues with starting and stopping nodes mid-test. + clearMapField(TCPEndpoint.class, null, "localEndpoints", this::endpointCreateByThisInstance); + Uninterruptibles.sleepUninterruptibly(2 * RMI_KEEPALIVE_TIME, TimeUnit.MILLISECONDS); // Double the keep-alive time to give Distributed GC some time to clean up } - private void clearMapField(Class clazz, Object instance, String mapName) - throws IllegalAccessException, NoSuchFieldException { - Field mapField = clazz.getDeclaredField(mapName); - mapField.setAccessible(true); - Map map = (Map) mapField.get(instance); - // Because multiple instances can be shutting down at once, - // synchronize on the map to avoid ConcurrentModificationException - synchronized (map) - { - for (Iterator> it = map.entrySet().iterator(); it.hasNext(); ) - { - it.next(); - it.remove(); - } - } + private boolean endpointCreateByThisInstance(Map.Entry> entry) + { + return entry.getValue() + .stream() + .anyMatch(ep -> ep.getServerSocketFactory() == this.serverSocketFactory && + ep.getClientSocketFactory() == this.clientSocketFactory); } } diff --git a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java new file mode 100644 index 0000000000..3e9798ea02 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java @@ -0,0 +1,226 @@ +/* + * 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.cassandra.distributed.shared; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.TimeUnit; +import java.util.function.BiConsumer; +import java.util.function.Consumer; +import java.util.function.Predicate; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import com.google.common.util.concurrent.Futures; + +import org.apache.cassandra.distributed.api.IInstance; +import org.apache.cassandra.distributed.api.NodeToolResult; + +import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly; + +public class ClusterUtils +{ + + /** + * Start the instance with the given System Properties, after the instance has started, the properties will be cleared. + */ + public static I start(I inst, Consumer fn) + { + return start(inst, (ignore, prop) -> fn.accept(prop)); + } + + /** + * Start the instance with the given System Properties, after the instance has started, the properties will be cleared. + */ + public static I start(I inst, BiConsumer fn) + { + try (WithProperties properties = new WithProperties()) + { + fn.accept(inst, properties); + inst.startup(); + return inst; + } + } + + /** + * Stop an instance in a blocking manner. + * + * The main difference between this and {@link IInstance#shutdown()} is that the wait on the future will catch + * the exceptions and throw as runtime. + */ + public static void stopUnchecked(IInstance i) + { + Futures.getUnchecked(i.shutdown()); + } + + public static List awaitRingStatus(IInstance instance, IInstance expectedInRing, String status) + { + return awaitInstanceMatching(instance, expectedInRing, d -> d.status.equals(status), + "Timeout waiting for " + expectedInRing + " to have status " + status); + } + + /** + * Wait for the ring to have the target instance with the provided state. + * + * @param instance instance to check on + * @param expectedInRing to look for + * @param state expected + * @return the ring + */ + public static List awaitRingState(IInstance instance, IInstance expectedInRing, String state) + { + return awaitInstanceMatching(instance, expectedInRing, d -> d.state.equals(state), + "Timeout waiting for " + expectedInRing + " to have state " + state); + } + + private static List awaitInstanceMatching(IInstance instance, + IInstance expectedInRing, + Predicate predicate, + String errorMessage) + { + return awaitRing(instance, + errorMessage, + ring -> ring.stream() + .filter(d -> d.address.equals(getBroadcastAddressHostString(expectedInRing))) + .anyMatch(predicate)); + } + + private static List awaitRing(IInstance src, String errorMessage, Predicate> fn) + { + List ring = null; + for (int i = 0; i < 100; i++) + { + ring = ring(src); + if (fn.test(ring)) + { + return ring; + } + sleepUninterruptibly(1, TimeUnit.SECONDS); + } + throw new AssertionError(errorMessage + "\n" + ring); + } + + /** + * Get the broadcast address host address only (ex. 127.0.0.1) + */ + private static String getBroadcastAddressHostString(IInstance target) + { + return target.config().broadcastAddress().getAddress().getHostAddress(); + } + + /** + * Get the ring from the perspective of the instance. + */ + public static List ring(IInstance inst) + { + NodeToolResult results = inst.nodetoolResult("ring"); + results.asserts().success(); + return parseRing(results.getStdout()); + } + + private static List parseRing(String str) + { + // 127.0.0.3 rack0 Up Normal 46.21 KB 100.00% -1 + // /127.0.0.1:7012 Unknown ? Normal ? 100.00% -3074457345618258603 + Pattern pattern = Pattern.compile("^(/?[0-9.:]+)\\s+(\\w+|\\?)\\s+(\\w+|\\?)\\s+(\\w+|\\?).*?(-?\\d+)\\s*$"); + List details = new ArrayList<>(); + String[] lines = str.split("\n"); + for (String line : lines) + { + Matcher matcher = pattern.matcher(line); + if (!matcher.find()) + { + continue; + } + details.add(new RingInstanceDetails(matcher.group(1), matcher.group(2), matcher.group(3), matcher.group(4), matcher.group(5))); + } + + return details; + } + + public static final class RingInstanceDetails + { + private final String address; + private final String rack; + private final String status; + private final String state; + private final String token; + + private RingInstanceDetails(String address, String rack, String status, String state, String token) + { + this.address = address; + this.rack = rack; + this.status = status; + this.state = state; + this.token = token; + } + + public String getAddress() + { + return address; + } + + public String getRack() + { + return rack; + } + + public String getStatus() + { + return status; + } + + public String getState() + { + return state; + } + + public String getToken() + { + return token; + } + + @Override + public boolean equals(Object o) + { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + RingInstanceDetails that = (RingInstanceDetails) o; + return Objects.equals(address, that.address) && + Objects.equals(rack, that.rack) && + Objects.equals(status, that.status) && + Objects.equals(state, that.state) && + Objects.equals(token, that.token); + } + + @Override + public int hashCode() + { + return Objects.hash(address, rack, status, state, token); + } + + public String toString() + { + return Arrays.asList(address, rack, status, state, token).toString(); + } + } + +} diff --git a/test/distributed/org/apache/cassandra/distributed/shared/WithProperties.java b/test/distributed/org/apache/cassandra/distributed/shared/WithProperties.java new file mode 100644 index 0000000000..a2aeae27ac --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/shared/WithProperties.java @@ -0,0 +1,112 @@ +/* + * 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.cassandra.distributed.shared; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.function.Supplier; + +import javax.annotation.Nullable; + +import com.google.common.base.Joiner; + +public final class WithProperties implements AutoCloseable +{ + private final List rollback = new ArrayList<>(); + + public WithProperties() + { + } + + public WithProperties with(String... kvs) + { + assert kvs.length % 2 == 0 : "Input must have an even amount of inputs but given " + kvs.length; + for (int i = 0; i <= kvs.length - 2; i = i + 2) + with(kvs[i], kvs[i + 1]); + return this; + } + + public WithProperties set(String prop, String value) + { + return set(prop, () -> System.setProperty(prop, value)); + } + + public WithProperties set(String prop, String... values) + { + return set(prop, Arrays.asList(values)); + } + + public WithProperties set(String prop, Collection values) + { + return set(prop, Joiner.on(",").join(values)); + } + + public WithProperties set(String prop, boolean value) + { + return set(prop, () -> System.setProperty(prop, convertToString(value))); + } + + public WithProperties set(String prop, long value) + { + return set(prop, () -> System.setProperty(prop, convertToString(value))); + } + + private void with(String key, String value) + { + String previous = System.setProperty(key, value); // checkstyle: suppress nearby 'blockSystemPropertyUsage' + rollback.add(previous == null ? () -> System.clearProperty(key) : () -> System.setProperty(key, previous)); // checkstyle: suppress nearby 'blockSystemPropertyUsage' + } + + private WithProperties set(String prop, Supplier prev) + { + String previous = convertToString(prev.get()); + rollback.add(previous == null ? () -> System.clearProperty(prop) : () -> System.setProperty(prop, previous)); + return this; + } + + @Override + public void close() + { + Collections.reverse(rollback); + rollback.forEach(Runnable::run); + rollback.clear(); + } + + public static String convertToString(@Nullable Object value) + { + if (value == null) + return null; + if (value instanceof String) + return (String) value; + if (value instanceof Boolean) + return Boolean.toString((Boolean) value); + if (value instanceof Long) + return Long.toString((Long) value); + if (value instanceof Integer) + return Integer.toString((Integer) value); + if (value instanceof Double) + return Double.toString((Double) value); + if (value instanceof Float) + return Float.toString((Float) value); + throw new IllegalArgumentException("Unknown type " + value.getClass()); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/ResourceLeakTest.java b/test/distributed/org/apache/cassandra/distributed/test/ResourceLeakTest.java index 177cbb9566..94b8bf8f73 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/ResourceLeakTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/ResourceLeakTest.java @@ -51,6 +51,7 @@ import static org.apache.cassandra.distributed.api.Feature.GOSSIP; import static org.apache.cassandra.distributed.api.Feature.JMX; import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL; import static org.apache.cassandra.distributed.api.Feature.NETWORK; +import static org.apache.cassandra.distributed.test.jmx.JMXGetterCheckTest.testAllValidGetters; import static org.hamcrest.Matchers.startsWith; /* Resource Leak Test - useful when tracking down issues with in-JVM framework cleanup. @@ -73,7 +74,7 @@ public class ResourceLeakTest extends TestBaseImpl final boolean dumpEveryLoop = false; // Dump heap & possibly files every loop final boolean dumpFileHandles = false; // Call lsof whenever dumping resources final boolean forceCollection = false; // Whether to explicitly force finalization/gc for smaller heap dumps - final long finalWaitMillis = 0l; // Number of millis to wait before final resource dump to give gc a chance + final long finalWaitMillis = 0L; // Number of millis to wait before final resource dump to give gc a chance static final SimpleDateFormat format = new SimpleDateFormat("yyyyMMddHHmmss"); static final String when = format.format(Date.from(Instant.now())); @@ -148,6 +149,30 @@ public class ResourceLeakTest extends TestBaseImpl } } + static void testJmx(Cluster cluster) + { + try + { + for (IInvokableInstance instance : cluster.get(1, cluster.size())) + { + IInstanceConfig config = instance.config(); + try (JMXConnector jmxc = JMXUtil.getJmxConnector(config, 5)) + { + MBeanServerConnection mbsc = jmxc.getMBeanServerConnection(); + // instances get their default domain set to their IP address, so us it + // to check that we are actually connecting to the correct instance + String defaultDomain = mbsc.getDefaultDomain(); + Assert.assertThat(defaultDomain, startsWith(JMXUtil.getJmxHost(config) + ":" + config.jmxPort())); + } + } + testAllValidGetters(cluster); + } + catch (Exception e) + { + throw new RuntimeException(e); + } + } + void doTest(int numClusterNodes, Consumer updater) throws Throwable { doTest(numClusterNodes, updater, ignored -> {}); @@ -231,27 +256,7 @@ public class ResourceLeakTest extends TestBaseImpl @Test public void looperJmxTest() throws Throwable { - doTest(1, config -> config.with(JMX), cluster -> { - // NOTE: At some point, the hostname of the broadcastAddress can be resolved - // and then the `getHostString`, which would otherwise return the IP address, - // starts returning `localhost` - use `.getAddress().getHostAddress()` to work around this. - for (IInvokableInstance instance:cluster.get(1, cluster.size())) - { - IInstanceConfig config = instance.config(); - try (JMXConnector jmxc = JMXUtil.getJmxConnector(config)) - { - MBeanServerConnection mbsc = jmxc.getMBeanServerConnection(); - // instances get their default domain set to their IP address, so us it - // to check that we are actually connecting to the correct instance - String defaultDomain = mbsc.getDefaultDomain(); - Assert.assertThat(defaultDomain, startsWith(JMXUtil.getJmxHost(config) + ":" + config.jmxPort())); - } - catch (IOException e) - { - throw new RuntimeException(e); - } - } - }); + doTest(2, config -> config.with(JMX), ResourceLeakTest::testJmx); if (forceCollection) { System.runFinalization(); @@ -264,7 +269,10 @@ public class ResourceLeakTest extends TestBaseImpl @Test public void looperEverythingTest() throws Throwable { - doTest(1, config -> config.with(Feature.values())); + doTest(2, config -> config.with(Feature.values()), + cluster -> { + testJmx(cluster); + }); if (forceCollection) { System.runFinalization(); diff --git a/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXFeatureTest.java b/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXFeatureTest.java index 1c38bd11e9..e227a24f8e 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXFeatureTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXFeatureTest.java @@ -18,7 +18,6 @@ package org.apache.cassandra.distributed.test.jmx; -import java.io.IOException; import java.util.HashSet; import java.util.Set; import javax.management.MBeanServerConnection; @@ -31,13 +30,20 @@ import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.Feature; import org.apache.cassandra.distributed.api.IInstanceConfig; import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.apache.cassandra.distributed.api.NodeToolResult; +import org.apache.cassandra.distributed.shared.ClusterUtils; import org.apache.cassandra.distributed.shared.JMXUtil; import org.apache.cassandra.distributed.test.TestBaseImpl; +import static org.apache.cassandra.distributed.test.jmx.JMXGetterCheckTest.testAllValidGetters; +import static org.hamcrest.Matchers.blankOrNullString; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.startsWith; public class JMXFeatureTest extends TestBaseImpl { + /** * Test the in-jvm dtest JMX feature. * - Create a cluster with multiple JMX servers, one per instance @@ -47,7 +53,7 @@ public class JMXFeatureTest extends TestBaseImpl * ports in addition to IP/Host for binding, but this version does not support that feature. Keeping the test name the same * so that it's consistent across versions. * - * @throws Exception + * @throws Exception it's a test that calls JMX endpoints - lots of different Jmx exceptions are possible */ @Test public void testMultipleNetworkInterfacesProvisioning() throws Exception @@ -59,18 +65,57 @@ public class JMXFeatureTest extends TestBaseImpl try (Cluster cluster = Cluster.build(2).withConfig(c -> c.with(Feature.values())).start()) { Set instancesContacted = new HashSet<>(); - for (IInvokableInstance instance : cluster.get(1, 2)) + for (IInvokableInstance instance : cluster) { testInstance(instancesContacted, instance); } Assert.assertEquals("Should have connected with both JMX instances.", 2, instancesContacted.size()); allInstances.addAll(instancesContacted); + // Make sure we actually exercise the mbeans by testing a bunch of getters. + // Without this it's possible for the test to pass as we don't touch any mBeans that we register. + testAllValidGetters(cluster); } } Assert.assertEquals("Each instance from each cluster should have been unique", iterations * 2, allInstances.size()); } - private void testInstance(Set instancesContacted, IInvokableInstance instance) throws IOException + @Test + public void testShutDownAndRestartInstances() throws Exception + { + HashSet instances = new HashSet<>(); + try (Cluster cluster = Cluster.build(2).withConfig(c -> c.with(Feature.values())).start()) + { + IInvokableInstance instanceToStop = cluster.get(1); + IInvokableInstance otherInstance = cluster.get(2); + testInstance(instances, cluster.get(1)); + testAllValidGetters(cluster); + ClusterUtils.stopUnchecked(instanceToStop); + // NOTE: This would previously fail because we cleared everything from the TCPEndpoint map in IsolatedJmx. + // Now, we only clear the endpoints related to that instance, which prevents this code from + // breaking with a `java.net.BindException: Address already in use (Bind failed)` + ClusterUtils.awaitRingStatus(otherInstance, instanceToStop, "Down"); + NodeToolResult statusResult = cluster.get(2).nodetoolResult("status"); + Assert.assertEquals(0, statusResult.getRc()); + Assert.assertThat(statusResult.getStderr(), is(blankOrNullString())); + Assert.assertThat(statusResult.getStdout(), containsString("DN 127.0.0.1")); + testInstance(instances, cluster.get(2)); + ClusterUtils.start(instanceToStop, props -> {}); + ClusterUtils.awaitRingState(otherInstance, instanceToStop, "Normal"); + ClusterUtils.awaitRingStatus(otherInstance, instanceToStop, "Up"); + statusResult = cluster.get(1).nodetoolResult("status"); + Assert.assertEquals(0, statusResult.getRc()); + Assert.assertThat(statusResult.getStderr(), is(blankOrNullString())); + Assert.assertThat(statusResult.getStdout(), containsString("UN 127.0.0.1")); + statusResult = cluster.get(2).nodetoolResult("status"); + Assert.assertEquals(0, statusResult.getRc()); + Assert.assertThat(statusResult.getStderr(), is(blankOrNullString())); + Assert.assertThat(statusResult.getStdout(), containsString("UN 127.0.0.1")); + testInstance(instances, cluster.get(1)); + testAllValidGetters(cluster); + } + } + + private void testInstance(Set instancesContacted, IInvokableInstance instance) { IInstanceConfig config = instance.config(); try (JMXConnector jmxc = JMXUtil.getJmxConnector(config)) @@ -82,5 +127,9 @@ public class JMXFeatureTest extends TestBaseImpl instancesContacted.add(defaultDomain); Assert.assertThat(defaultDomain, startsWith(JMXUtil.getJmxHost(config) + ":" + config.jmxPort())); } + catch (Throwable t) + { + throw new RuntimeException("Could not connect to JMX", t); + } } } diff --git a/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXGetterCheckTest.java b/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXGetterCheckTest.java index a3f74df60d..bbcfa52ebc 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXGetterCheckTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXGetterCheckTest.java @@ -29,44 +29,68 @@ import javax.management.MBeanOperationInfo; import javax.management.MBeanServerConnection; import javax.management.ObjectName; import javax.management.remote.JMXConnector; -import javax.management.remote.JMXConnectorFactory; -import javax.management.remote.JMXServiceURL; import com.google.common.collect.ImmutableSet; import org.junit.Test; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.distributed.api.IInstanceConfig; import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.apache.cassandra.distributed.shared.JMXUtil; import org.apache.cassandra.distributed.test.TestBaseImpl; public class JMXGetterCheckTest extends TestBaseImpl { private static final Set IGNORE_ATTRIBUTES = ImmutableSet.of( - "org.apache.cassandra.net:type=MessagingService:BackPressurePerHost" // throws unsupported saying the feature was removed... dropped in CASSANDRA-15375 + "org.apache.cassandra.net:type=MessagingService:BackPressurePerHost", // throws unsupported saying the feature was removed... dropped in CASSANDRA-15375 + "org.apache.cassandra.db:type=StorageService:MaxNativeProtocolVersion" // Throws NPE ); private static final Set IGNORE_OPERATIONS = ImmutableSet.of( "org.apache.cassandra.db:type=StorageService:stopDaemon", // halts the instance, which then causes the JVM to exit "org.apache.cassandra.db:type=StorageService:drain", // don't drain, it stops things which can cause other APIs to be unstable as we are in a stopped state "org.apache.cassandra.db:type=StorageService:stopGossiping", // if we stop gossip this can cause other issues, so avoid "org.apache.cassandra.db:type=StorageService:resetLocalSchema", // this will fail when there are no other nodes which can serve schema - "org.apache.cassandra.db:type=HintedHandoffManager:listEndpointsPendingHints", // this will fail because it only exists to match an old, deprecated mbean and just throws an UnsportedOperationException - "org.apache.cassandra.db:type=StorageService:decommission" // Don't decommission nodes! Note that in future versions of C* this is unnecessary because decommission takes an argument. + "org.apache.cassandra.db:type=StorageService:joinRing", // Causes bootstrapping errors + "org.apache.cassandra.db:type=StorageService:clearConnectionHistory", // Throws a NullPointerException + "org.apache.cassandra.db:type=StorageService:startGossiping", // causes multiple loops to fail + "org.apache.cassandra.db:type=StorageService:startNativeTransport", // causes multiple loops to fail + "org.apache.cassandra.db:type=Tables,keyspace=system,table=local:loadNewSSTables", // Shouldn't attempt to load SSTables as sometimes the temp directories don't work + "org.apache.cassandra.db:type=HintedHandoffManager:listEndpointsPendingHints", // hard-coded to throw UnsupportedOperationException + "org.apache.cassandra.db:type=StorageService:startRPCServer", // In looped tests this can sometimes fail + "org.apache.cassandra.db:type=StorageService:decommission" // shutting down the node and decommissioning it is bad - in later versions, it takes a parameter and wouldn't be run in this test ); - public static final String JMX_SERVICE_URL_FMT = "service:jmx:rmi:///jndi/rmi://%s:%d/jmxrmi"; - @Test public void testGetters() throws Exception { - try (Cluster cluster = Cluster.build(1).withConfig(c -> c.with(Feature.values())).start()) + for (int i=0; i < 2; i++) { - IInvokableInstance instance = cluster.get(1); + try (Cluster cluster = Cluster.build(1).withConfig(c -> c.with(Feature.values())).start()) + { + testAllValidGetters(cluster); + } + } + } - String jmxHost = instance.config().broadcastAddress().getAddress().getHostAddress(); - String url = String.format(JMX_SERVICE_URL_FMT, jmxHost, instance.config().jmxPort()); + /** + * Tests JMX getters and operations. + * Useful for more than just testing getters, it also is used in JMXFeatureTest + * to make sure we've touched the complete JMX code path. + * @param cluster the cluster to test + * @throws Exception several kinds of exceptions can be thrown, mostly from JMX infrastructure issues. + */ + public static void testAllValidGetters(Cluster cluster) throws Exception + { + for (IInvokableInstance instance: cluster) + { + if (instance.isShutdown()) + { + continue; + } + IInstanceConfig config = instance.config(); List errors = new ArrayList<>(); - try (JMXConnector jmxc = JMXConnectorFactory.connect(new JMXServiceURL(url), null)) + try (JMXConnector jmxc = JMXUtil.getJmxConnector(config)) { MBeanServerConnection mbsc = jmxc.getMBeanServerConnection(); Set metricNames = new TreeSet<>(mbsc.queryNames(null, null));