diff --git a/build.xml b/build.xml index e56c1f8e8a..da21569de7 100644 --- a/build.xml +++ b/build.xml @@ -152,7 +152,7 @@ - + @@ -217,6 +217,7 @@ --add-exports java.management.rmi/com.sun.jmx.remote.internal.rmi=ALL-UNNAMED --add-exports java.rmi/sun.rmi.registry=ALL-UNNAMED --add-exports java.rmi/sun.rmi.server=ALL-UNNAMED + --add-exports java.rmi/sun.rmi.transport.tcp=ALL-UNNAMED --add-exports java.sql/java.sql=ALL-UNNAMED --add-opens java.base/java.lang.module=ALL-UNNAMED @@ -228,6 +229,7 @@ --add-opens java.base/jdk.internal.module=ALL-UNNAMED --add-opens java.base/jdk.internal.util.jar=ALL-UNNAMED --add-opens jdk.management/com.sun.management.internal=ALL-UNNAMED + @@ -268,7 +270,7 @@ - + @@ -1441,6 +1443,7 @@ + @@ -2166,7 +2169,7 @@ - ]]> diff --git a/relocate-dependencies.pom b/relocate-dependencies.pom index 07728dd405..96d8d16ee9 100644 --- a/relocate-dependencies.pom +++ b/relocate-dependencies.pom @@ -67,7 +67,7 @@ org.apache.maven.plugins maven-shade-plugin - 3.2.1 + 3.4.1 false diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java index 3e45ebc3ed..75e36b4693 100644 --- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java +++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java @@ -88,6 +88,19 @@ public enum CassandraRelevantProperties */ COM_SUN_MANAGEMENT_JMXREMOTE_AUTHENTICATE ("com.sun.management.jmxremote.authenticate"), + + /** + * Controls the JMX server threadpool keap-alive time. + * Should only be set by in-jvm dtests. + */ + SUN_RMI_TRANSPORT_TCP_THREADKEEPALIVETIME("sun.rmi.transport.tcp.threadKeepAliveTime"), + + /** + * Controls the distributed garbage collector lease time for JMX objects. + * Should only be set by in-jvm dtests. + */ + JAVA_RMI_DGC_LEASE_VALUE_IN_JVM_DTEST("java.rmi.dgc.leaseValue"), + /** * The port number to which the RMI connector will be bound - com.sun.management.jmxremote.rmi.port. * An Integer object that represents the value of the second argument is returned @@ -226,6 +239,9 @@ public enum CassandraRelevantProperties /** what class to use for mbean registeration */ MBEAN_REGISTRATION_CLASS("org.apache.cassandra.mbean_registration_class"), + /** This property indicates if the code is running under the in-jvm dtest framework */ + DTEST_IS_IN_JVM_DTEST("org.apache.cassandra.dtest.is_in_jvm_dtest"), + BATCH_COMMIT_LOG_SYNC_INTERVAL("cassandra.batch_commitlog_sync_interval_millis", "1000"), SYSTEM_AUTH_DEFAULT_RF("cassandra.system_auth.default_rf", "1"), diff --git a/src/java/org/apache/cassandra/service/GCInspector.java b/src/java/org/apache/cassandra/service/GCInspector.java index c290d9e983..8f922156ca 100644 --- a/src/java/org/apache/cassandra/service/GCInspector.java +++ b/src/java/org/apache/cassandra/service/GCInspector.java @@ -148,18 +148,16 @@ public class GCInspector implements NotificationListener, GCInspectorMXBean public GCInspector() { - MBeanServer mbs = ManagementFactory.getPlatformMBeanServer(); - try { ObjectName gcName = new ObjectName(ManagementFactory.GARBAGE_COLLECTOR_MXBEAN_DOMAIN_TYPE + ",*"); - for (ObjectName name : mbs.queryNames(gcName, null)) + for (ObjectName name : MBeanWrapper.instance.queryNames(gcName, null)) { - GarbageCollectorMXBean gc = ManagementFactory.newPlatformMXBeanProxy(mbs, name.getCanonicalName(), GarbageCollectorMXBean.class); + GarbageCollectorMXBean gc = ManagementFactory.newPlatformMXBeanProxy(MBeanWrapper.instance.getMBeanServer(), name.getCanonicalName(), GarbageCollectorMXBean.class); gcStates.put(gc.getName(), new GCState(gc, assumeGCIsPartiallyConcurrent(gc), assumeGCIsOldGen(gc))); } ObjectName me = new ObjectName(MBEAN_NAME); - if (!mbs.isRegistered(me)) + if (!MBeanWrapper.instance.isRegistered(me)) MBeanWrapper.instance.registerMBean(this, new ObjectName(MBEAN_NAME)); } catch (MalformedObjectNameException | IOException e) diff --git a/src/java/org/apache/cassandra/utils/JMXServerUtils.java b/src/java/org/apache/cassandra/utils/JMXServerUtils.java index 5557fda345..49ff5a07d2 100644 --- a/src/java/org/apache/cassandra/utils/JMXServerUtils.java +++ b/src/java/org/apache/cassandra/utils/JMXServerUtils.java @@ -29,12 +29,14 @@ import java.net.Inet6Address; import java.net.InetAddress; import java.rmi.AccessException; import java.rmi.AlreadyBoundException; +import java.rmi.NoSuchObjectException; import java.rmi.NotBoundException; import java.rmi.Remote; import java.rmi.RemoteException; import java.rmi.registry.Registry; import java.rmi.server.RMIClientSocketFactory; import java.rmi.server.RMIServerSocketFactory; +import java.rmi.server.UnicastRemoteObject; import java.util.Arrays; import java.util.HashMap; import java.util.Map; @@ -252,7 +254,8 @@ public class JMXServerUtils return env; } - private static void logJmxServiceUrl(InetAddress serverAddress, int port) + @VisibleForTesting + public static void logJmxServiceUrl(InetAddress serverAddress, int port) { String urlTemplate = "service:jmx:rmi://%1$s/jndi/rmi://%1$s:%2$d/jmxrmi"; String hostName; @@ -325,11 +328,11 @@ public class JMXServerUtils * Better to use the internal API than re-invent the wheel. */ @SuppressWarnings("restriction") - private static class JmxRegistry extends sun.rmi.registry.RegistryImpl { + public static class JmxRegistry extends sun.rmi.registry.RegistryImpl { private final String lookupName; private Remote remoteServerStub; - JmxRegistry(final int port, + public JmxRegistry(final int port, final RMIClientSocketFactory csf, RMIServerSocketFactory ssf, final String lookupName) throws RemoteException { @@ -362,5 +365,24 @@ public class JMXServerUtils public void setRemoteServerStub(Remote remoteServerStub) { this.remoteServerStub = remoteServerStub; } + + /** + * Closes the underlying JMX registry by unexporting this instance. + * There is no reason to do this except for in-jvm dtests where we need + * to stop the registry, so we can start with a clean slate for future cluster + * builds, and the superclass never expects to be shut down and therefore doesn't + * handle this edge case at all. + */ + @VisibleForTesting + public void close() { + try + { + UnicastRemoteObject.unexportObject(this, true); + } + catch (NoSuchObjectException ignored) + { + // Ignore if it's already unexported + } + } } } diff --git a/src/java/org/apache/cassandra/utils/MBeanWrapper.java b/src/java/org/apache/cassandra/utils/MBeanWrapper.java index f8bc439d7b..0ef342da30 100644 --- a/src/java/org/apache/cassandra/utils/MBeanWrapper.java +++ b/src/java/org/apache/cassandra/utils/MBeanWrapper.java @@ -19,15 +19,21 @@ package org.apache.cassandra.utils; import java.lang.management.ManagementFactory; +import java.util.Collections; +import java.util.Set; +import java.util.UUID; import java.util.function.Consumer; import javax.management.MBeanServer; +import javax.management.MBeanServerFactory; import javax.management.MalformedObjectNameException; import javax.management.ObjectName; +import javax.management.QueryExp; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import static org.apache.cassandra.config.CassandraRelevantProperties.IS_DISABLED_MBEAN_REGISTRATION; +import static org.apache.cassandra.config.CassandraRelevantProperties.ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION; +import static org.apache.cassandra.config.CassandraRelevantProperties.DTEST_IS_IN_JVM_DTEST; import static org.apache.cassandra.config.CassandraRelevantProperties.MBEAN_REGISTRATION_CLASS; /** @@ -36,23 +42,46 @@ import static org.apache.cassandra.config.CassandraRelevantProperties.MBEAN_REGI */ public interface MBeanWrapper { - static final Logger logger = LoggerFactory.getLogger(MBeanWrapper.class); + Logger logger = LoggerFactory.getLogger(MBeanWrapper.class); - static final MBeanWrapper instance = create(); + MBeanWrapper instance = create(); static MBeanWrapper create() { - if (IS_DISABLED_MBEAN_REGISTRATION.getBoolean()) + // If we're running in the in-jvm dtest environment, always use the delegating + // mbean wrapper even if we start off with no-op, so it can be switched later + if (DTEST_IS_IN_JVM_DTEST.getBoolean()) + { + return new DelegatingMbeanWrapper(getMBeanWrapper()); + } + + return getMBeanWrapper(); + } + + static MBeanWrapper getMBeanWrapper() + { + if (ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION.getBoolean()) + { return new NoOpMBeanWrapper(); + } String klass = MBEAN_REGISTRATION_CLASS.getString(); if (klass == null) - return new PlatformMBeanWrapper(); + { + if (DTEST_IS_IN_JVM_DTEST.getBoolean()) + { + return new NoOpMBeanWrapper(); + } + else + { + return new PlatformMBeanWrapper(); + } + } return FBUtilities.construct(klass, "mbean"); } // Passing true for graceful will log exceptions instead of rethrowing them - public void registerMBean(Object obj, ObjectName mbeanName, OnException onException); + void registerMBean(Object obj, ObjectName mbeanName, OnException onException); default void registerMBean(Object obj, ObjectName mbeanName) { registerMBean(obj, mbeanName, OnException.THROW); @@ -62,7 +91,9 @@ public interface MBeanWrapper { ObjectName name = create(mbeanName, onException); if (name == null) + { return; + } registerMBean(obj, name, onException); } default void registerMBean(Object obj, String mbeanName) @@ -70,7 +101,7 @@ public interface MBeanWrapper registerMBean(obj, mbeanName, OnException.THROW); } - public boolean isRegistered(ObjectName mbeanName, OnException onException); + boolean isRegistered(ObjectName mbeanName, OnException onException); default boolean isRegistered(ObjectName mbeanName) { return isRegistered(mbeanName, OnException.THROW); @@ -80,7 +111,9 @@ public interface MBeanWrapper { ObjectName name = create(mbeanName, onException); if (name == null) + { return false; + } return isRegistered(name, onException); } default boolean isRegistered(String mbeanName) @@ -88,7 +121,7 @@ public interface MBeanWrapper return isRegistered(mbeanName, OnException.THROW); } - public void unregisterMBean(ObjectName mbeanName, OnException onException); + void unregisterMBean(ObjectName mbeanName, OnException onException); default void unregisterMBean(ObjectName mbeanName) { unregisterMBean(mbeanName, OnException.THROW); @@ -98,7 +131,9 @@ public interface MBeanWrapper { ObjectName name = create(mbeanName, onException); if (name == null) + { return; + } unregisterMBean(name, onException); } default void unregisterMBean(String mbeanName) @@ -119,7 +154,11 @@ public interface MBeanWrapper } } - static class NoOpMBeanWrapper implements MBeanWrapper + Set queryNames(ObjectName name, QueryExp query); + + MBeanServer getMBeanServer(); + + class NoOpMBeanWrapper implements MBeanWrapper { public void registerMBean(Object obj, ObjectName mbeanName, OnException onException) {} public void registerMBean(Object obj, String mbeanName, OnException onException) {} @@ -127,9 +166,11 @@ public interface MBeanWrapper public boolean isRegistered(String mbeanName, OnException onException) { return false; } public void unregisterMBean(ObjectName mbeanName, OnException onException) {} public void unregisterMBean(String mbeanName, OnException onException) {} + public Set queryNames(ObjectName name, QueryExp query) {return Collections.emptySet(); } + public MBeanServer getMBeanServer() { return null; } } - static class PlatformMBeanWrapper implements MBeanWrapper + class PlatformMBeanWrapper implements MBeanWrapper { private final MBeanServer mbs = ManagementFactory.getPlatformMBeanServer(); public void registerMBean(Object obj, ObjectName mbeanName, OnException onException) @@ -168,9 +209,158 @@ public interface MBeanWrapper onException.handler.accept(e); } } + + public Set queryNames(ObjectName name, QueryExp query) + { + return mbs.queryNames(name, query); + } + + public MBeanServer getMBeanServer() + { + return mbs; + } } - public enum OnException + class InstanceMBeanWrapper implements MBeanWrapper + { + private MBeanServer mbs; + public final UUID id = UUID.randomUUID(); + + public InstanceMBeanWrapper(String hostname) + { + mbs = MBeanServerFactory.createMBeanServer(hostname + "-" + id); + } + + public void registerMBean(Object obj, ObjectName mbeanName, OnException onException) + { + try + { + mbs.registerMBean(obj, mbeanName); + } + catch (Exception e) + { + onException.handler.accept(e); + } + } + + public boolean isRegistered(ObjectName mbeanName, OnException onException) + { + try + { + return mbs.isRegistered(mbeanName); + } + catch (Exception e) + { + onException.handler.accept(e); + } + return false; + } + + public void unregisterMBean(ObjectName mbeanName, OnException onException) + { + try + { + mbs.unregisterMBean(mbeanName); + } + catch (Exception e) + { + onException.handler.accept(e); + } + } + + public Set queryNames(ObjectName name, QueryExp query) + { + return mbs.queryNames(name, query); + } + + public MBeanServer getMBeanServer() + { + return mbs; + } + + public void close() { + mbs.queryNames(null, null).forEach(name -> { + try { + if (!name.getCanonicalName().contains("MBeanServerDelegate")) + { + mbs.unregisterMBean(name); + } + } catch (Throwable e) { + logger.debug("Could not unregister mbean {}", name.getCanonicalName()); + } + }); + MBeanServerFactory.releaseMBeanServer(mbs); + mbs = null; + } + } + + class DelegatingMbeanWrapper implements MBeanWrapper + { + MBeanWrapper delegate; + + public DelegatingMbeanWrapper(MBeanWrapper mBeanWrapper) + { + delegate = mBeanWrapper; + } + + public void setDelegate(MBeanWrapper wrapper) { + delegate = wrapper; + } + + public MBeanWrapper getDelegate() + { + return delegate; + } + + public void registerMBean(Object obj, ObjectName mbeanName, OnException onException) + { + try + { + delegate.registerMBean(obj, mbeanName); + } + catch (Exception e) + { + onException.handler.accept(e); + } + } + + public boolean isRegistered(ObjectName mbeanName, OnException onException) + { + try + { + return delegate.isRegistered(mbeanName); + } + catch (Exception e) + { + onException.handler.accept(e); + } + return false; + } + + public void unregisterMBean(ObjectName mbeanName, OnException onException) + { + try + { + delegate.unregisterMBean(mbeanName); + } + catch (Exception e) + { + onException.handler.accept(e); + } + } + + public Set queryNames(ObjectName name, QueryExp query) + { + return delegate.queryNames(name, query); + } + + public MBeanServer getMBeanServer() + { + return delegate.getMBeanServer(); + } + } + + enum OnException { THROW(e -> { throw new RuntimeException(e); }), LOG(e -> { logger.error("Error in MBean wrapper: ", e); }), diff --git a/src/java/org/apache/cassandra/utils/RMIClientSocketFactoryImpl.java b/src/java/org/apache/cassandra/utils/RMIClientSocketFactoryImpl.java new file mode 100644 index 0000000000..62ab88fb92 --- /dev/null +++ b/src/java/org/apache/cassandra/utils/RMIClientSocketFactoryImpl.java @@ -0,0 +1,62 @@ +/* + * 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.io.IOException; +import java.io.Serializable; +import java.net.InetAddress; +import java.net.Socket; +import java.rmi.server.RMIClientSocketFactory; +import java.util.Objects; + +/** + * This class is used to override the local address the JMX client calculates when trying to connect, + * which can otherwise be influenced by the system property "java.rmi.server.hostname" in strange and + * unpredictable ways. + */ +public class RMIClientSocketFactoryImpl implements RMIClientSocketFactory, Serializable +{ + private final InetAddress localAddress; + + public RMIClientSocketFactoryImpl(InetAddress localAddress) + { + this.localAddress = localAddress; + } + + @Override + public Socket createSocket(String host, int port) throws IOException + { + return new Socket(localAddress, port); + } + + @Override + public boolean equals(Object o) + { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + RMIClientSocketFactoryImpl that = (RMIClientSocketFactoryImpl) o; + return Objects.equals(localAddress, that.localAddress); + } + + @Override + public int hashCode() + { + return Objects.hash(localAddress); + } +} 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..bd605db3bc --- /dev/null +++ b/src/java/org/apache/cassandra/utils/ReflectionUtils.java @@ -0,0 +1,55 @@ +/* + * 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; + +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; + } + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java index 7d333726e5..47371772b9 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java @@ -186,6 +186,8 @@ public abstract class AbstractCluster implements ICluster implements ICluster get(int... nodes) + { + if (nodes == null || nodes.length == 0) + throw new IllegalArgumentException("No nodes provided"); + List list = new ArrayList<>(nodes.length); + for (int i : nodes) + list.add(get(i)); + return list; + } + public I getFirstRunningInstance() { return stream().filter(i -> !i.isShutdown()).findFirst().orElseThrow( diff --git a/test/distributed/org/apache/cassandra/distributed/impl/CollectingRMIServerSocketFactoryImpl.java b/test/distributed/org/apache/cassandra/distributed/impl/CollectingRMIServerSocketFactoryImpl.java new file mode 100644 index 0000000000..5e67eafef6 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/impl/CollectingRMIServerSocketFactoryImpl.java @@ -0,0 +1,87 @@ +/* + * 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.impl; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.SocketException; +import java.rmi.server.RMIServerSocketFactory; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; +import javax.net.ServerSocketFactory; + + +/** + * This class is used to keep track of RMI servers created during a cluster creation so we can + * later close the sockets, which would otherwise be left with a thread running waiting for + * connections that would never show up as the server was otherwise closed. + */ +class CollectingRMIServerSocketFactoryImpl implements RMIServerSocketFactory +{ + private final InetAddress bindAddress; + List sockets = new ArrayList<>(); + + public CollectingRMIServerSocketFactoryImpl(InetAddress bindAddress) + { + this.bindAddress = bindAddress; + } + + @Override + public ServerSocket createServerSocket(int pPort) throws IOException + { + ServerSocket result = ServerSocketFactory.getDefault().createServerSocket(pPort, 0, bindAddress); + try + { + result.setReuseAddress(true); + } + catch (SocketException e) + { + result.close(); + throw e; + } + sockets.add(result); + return result; + } + + + public void close() throws IOException + { + for (ServerSocket socket : sockets) + { + socket.close(); + } + } + + @Override + public boolean equals(Object o) + { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + CollectingRMIServerSocketFactoryImpl that = (CollectingRMIServerSocketFactoryImpl) o; + return Objects.equals(bindAddress, that.bindAddress); + } + + @Override + public int hashCode() + { + return Objects.hash(bindAddress); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/impl/INodeProvisionStrategy.java b/test/distributed/org/apache/cassandra/distributed/impl/INodeProvisionStrategy.java index 1ec844e533..99ef272de0 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/INodeProvisionStrategy.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/INodeProvisionStrategy.java @@ -56,6 +56,11 @@ public interface INodeProvisionStrategy { return 9041 + nodeNum; } + + public int jmxPort(int nodeNum) + { + return 7199 + nodeNum; + } }; } }, @@ -89,6 +94,11 @@ public interface INodeProvisionStrategy { return 9042; } + + public int jmxPort(int nodeNum) + { + return 7199; + } }; } }; @@ -100,4 +110,5 @@ public interface INodeProvisionStrategy abstract String ipAddress(int nodeNum); abstract int storagePort(int nodeNum); abstract int nativeTransportPort(int nodeNum); + abstract int jmxPort(int nodeNum); } diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java index d4cb1cb9ed..9975aeb9eb 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java @@ -42,6 +42,8 @@ import java.util.stream.Stream; import javax.management.ListenerNotFoundException; import javax.management.Notification; import javax.management.NotificationListener; +import javax.management.remote.JMXConnectorServer; +import javax.management.remote.rmi.RMIJRMPServerImpl; import com.google.common.annotations.VisibleForTesting; import com.google.common.util.concurrent.Uninterruptibles; @@ -143,7 +145,10 @@ import org.apache.cassandra.utils.Closeable; import org.apache.cassandra.utils.DiagnosticSnapshotService; import org.apache.cassandra.utils.ExecutorUtils; import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.JMXServerUtils; import org.apache.cassandra.utils.JVMStabilityInspector; +import org.apache.cassandra.utils.MBeanWrapper; +import org.apache.cassandra.utils.RMIClientSocketFactoryImpl; import org.apache.cassandra.utils.Throwables; import org.apache.cassandra.utils.concurrent.Ref; import org.apache.cassandra.utils.memory.BufferPools; @@ -153,6 +158,7 @@ import static java.util.concurrent.TimeUnit.MINUTES; import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory; import static org.apache.cassandra.distributed.api.Feature.BLANK_GOSSIP; 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.impl.DistributedTestSnitch.fromCassandraInetAddressAndPort; @@ -165,11 +171,19 @@ import static org.apache.cassandra.utils.Clock.Global.nanoTime; */ public class Instance extends IsolatedExecutor implements IInvokableInstance { + private static final int RMI_KEEPALIVE_TIME = 1000; private Logger inInstancelogger; // Defer creation until running in the instance context public final IInstanceConfig config; private volatile boolean initialized = false; private volatile boolean internodeMessagingStarted = false; private final AtomicLong startedAt = new AtomicLong(); + private JMXConnectorServer jmxConnectorServer; + private JMXServerUtils.JmxRegistry registry; + private RMIJRMPServerImpl jmxRmiServer; + private MBeanWrapper.InstanceMBeanWrapper wrapper; + private RMIClientSocketFactoryImpl clientSocketFactory; + private CollectingRMIServerSocketFactoryImpl serverSocketFactory; + private IsolatedJmx isolatedJmx; @Deprecated Instance(IInstanceConfig config, ClassLoader classLoader) @@ -589,6 +603,9 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance config.networkTopology(), config.broadcastAddress()); DistributedTestSnitch.assign(config.networkTopology()); + if (config.has(JMX)) + startJmx(); + DatabaseDescriptor.daemonInitialization(); FileUtils.setFSErrorHandler(new DefaultFSErrorHandler()); DatabaseDescriptor.createAllDirectories(); @@ -738,6 +755,20 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance initialized = true; } + private void startJmx() + { + this.isolatedJmx = new IsolatedJmx(this, inInstancelogger); + isolatedJmx.startJmx(); + } + + private void stopJmx() throws IllegalAccessException, NoSuchFieldException, InterruptedException + { + if (config.has(JMX)) + { + isolatedJmx.stopJmx(); + } + } + // Update the messaging versions for all instances // that have initialized their configurations. private static void propagateMessagingVersions(ICluster cluster) @@ -879,6 +910,8 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance // ScheduledExecutors shuts down after MessagingService, as MessagingService may issue tasks to it. error = parallelRun(error, executor, () -> ScheduledExecutors.shutdownNowAndWait(1L, MINUTES)); + + error = parallelRun(error, executor, this::stopJmx); // Make sure any shutdown hooks registered for DeleteOnExit are released to prevent // references to the instance class loaders from being held diff --git a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java index 92c56d6267..3c515d57f0 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java @@ -41,6 +41,8 @@ import org.apache.cassandra.locator.SimpleSeedProvider; public class InstanceConfig implements IInstanceConfig { public final int num; + private final int jmxPort; + public int num() { return num; } private final NetworkTopology networkTopology; @@ -71,7 +73,8 @@ public class InstanceConfig implements IInstanceConfig String cdc_raw_directory, Collection initial_token, int storage_port, - int native_transport_port) + int native_transport_port, + int jmx_port) { this.num = num; this.networkTopology = networkTopology; @@ -113,6 +116,7 @@ public class InstanceConfig implements IInstanceConfig // legacy parameters .forceSet("commitlog_sync_batch_window_in_ms", "1"); this.featureFlags = EnumSet.noneOf(Feature.class); + this.jmxPort = jmx_port; } private InstanceConfig(InstanceConfig copy) @@ -124,6 +128,7 @@ public class InstanceConfig implements IInstanceConfig this.hostId = copy.hostId; this.featureFlags = copy.featureFlags; this.broadcastAddressAndPort = copy.broadcastAddressAndPort; + this.jmxPort = copy.jmxPort; } @Override @@ -168,6 +173,12 @@ public class InstanceConfig implements IInstanceConfig return networkTopology().localDC(broadcastAddress()); } + @Override + public int jmxPort() + { + return this.jmxPort; + } + public InstanceConfig with(Feature featureFlag) { featureFlags.add(featureFlag); @@ -267,7 +278,8 @@ public class InstanceConfig implements IInstanceConfig String.format("%s/node%d/cdc", root, nodeNum), tokens, provisionStrategy.storagePort(nodeNum), - provisionStrategy.nativeTransportPort(nodeNum)); + provisionStrategy.nativeTransportPort(nodeNum), + provisionStrategy.jmxPort(nodeNum)); } private static String[] datadirs(int datadirCount, Path root, int nodeNum) diff --git a/test/distributed/org/apache/cassandra/distributed/impl/IsolatedJmx.java b/test/distributed/org/apache/cassandra/distributed/impl/IsolatedJmx.java new file mode 100644 index 0000000000..e19e29fdb3 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/impl/IsolatedJmx.java @@ -0,0 +1,230 @@ +/* + * 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.impl; + +import java.lang.reflect.Field; +import java.net.InetAddress; +import java.net.MalformedURLException; +import java.util.HashMap; +import java.util.Iterator; +import java.util.Map; + +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; +import javax.management.remote.rmi.RMIJRMPServerImpl; + +import org.slf4j.Logger; + +import org.apache.cassandra.distributed.api.IInstance; +import org.apache.cassandra.distributed.api.IInstanceConfig; +import org.apache.cassandra.utils.JMXServerUtils; +import org.apache.cassandra.utils.MBeanWrapper; +import org.apache.cassandra.utils.RMIClientSocketFactoryImpl; +import org.apache.cassandra.utils.ReflectionUtils; +import sun.rmi.transport.tcp.TCPEndpoint; + +import static org.apache.cassandra.config.CassandraRelevantProperties.JAVA_RMI_DGC_LEASE_VALUE_IN_JVM_DTEST; +import static org.apache.cassandra.config.CassandraRelevantProperties.ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION; +import static org.apache.cassandra.config.CassandraRelevantProperties.SUN_RMI_TRANSPORT_TCP_THREADKEEPALIVETIME; +import static org.apache.cassandra.distributed.api.Feature.JMX; + +public class IsolatedJmx +{ + private static final int RMI_KEEPALIVE_TIME = 1000; + + private JMXConnectorServer jmxConnectorServer; + private JMXServerUtils.JmxRegistry registry; + private RMIJRMPServerImpl jmxRmiServer; + private MBeanWrapper.InstanceMBeanWrapper wrapper; + private RMIClientSocketFactoryImpl clientSocketFactory; + private CollectingRMIServerSocketFactoryImpl serverSocketFactory; + private Logger inInstancelogger; + private IInstanceConfig config; + + public IsolatedJmx(IInstance instance, Logger inInstanceLogger) { + this.config = instance.config(); + this.inInstancelogger = inInstanceLogger; + } + + public void startJmx() { + try + { + // Several RMI threads hold references to in-jvm dtest objects, and are, by default, kept + // alive for long enough (minutes) to keep classloaders from being collected. + // Set these two system properties to a low value to allow cleanup to occur fast enough + // for GC to collect our classloaders. + JAVA_RMI_DGC_LEASE_VALUE_IN_JVM_DTEST.setInt(RMI_KEEPALIVE_TIME); + SUN_RMI_TRANSPORT_TCP_THREADKEEPALIVETIME.setInt(RMI_KEEPALIVE_TIME); + ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION.setBoolean(false); + InetAddress addr = config.broadcastAddress().getAddress(); + + int jmxPort = config.jmxPort(); + + String hostname = addr.getHostAddress(); + wrapper = new MBeanWrapper.InstanceMBeanWrapper(hostname + ":" + jmxPort); + ((MBeanWrapper.DelegatingMbeanWrapper) MBeanWrapper.instance).setDelegate(wrapper); + Map env = new HashMap<>(); + + serverSocketFactory = new CollectingRMIServerSocketFactoryImpl(addr); + env.put(RMIConnectorServer.RMI_SERVER_SOCKET_FACTORY_ATTRIBUTE, + serverSocketFactory); + clientSocketFactory = new RMIClientSocketFactoryImpl(addr); + env.put(RMIConnectorServer.RMI_CLIENT_SOCKET_FACTORY_ATTRIBUTE, + clientSocketFactory); + + // configure the RMI registry + registry = new JMXServerUtils.JmxRegistry(jmxPort, + clientSocketFactory, + serverSocketFactory, + "jmxrmi"); + + // Mark the JMX server as a permanently exported object. This allows the JVM to exit with the + // server running and also exempts it from the distributed GC scheduler which otherwise would + // potentially attempt a full GC every `sun.rmi.dgc.server.gcInterval` millis (default is 3600000ms) + // For more background see: + // - CASSANDRA-2967 + // - https://www.jclarity.com/2015/01/27/rmi-system-gc-unplugged/ + // - https://bugs.openjdk.java.net/browse/JDK-6760712 + env.put("jmx.remote.x.daemon", "true"); + + // Set the port used to create subsequent connections to exported objects over RMI. This simplifies + // configuration in firewalled environments, but it can't be used in conjuction with SSL sockets. + // See: CASSANDRA-7087 + int rmiPort = config.jmxPort(); + + // We create the underlying RMIJRMPServerImpl so that we can manually bind it to the registry, + // rather then specifying a binding address in the JMXServiceURL and letting it be done automatically + // when the server is started. The reason for this is that if the registry is configured with SSL + // sockets, the JMXConnectorServer acts as its client during the binding which means it needs to + // have a truststore configured which contains the registry's certificate. Manually binding removes + // this problem. + // See CASSANDRA-12109. + jmxRmiServer = new RMIJRMPServerImpl(rmiPort, clientSocketFactory, serverSocketFactory, + env); + JMXServiceURL serviceURL = new JMXServiceURL("rmi", hostname, rmiPort); + jmxConnectorServer = new RMIConnectorServer(serviceURL, env, jmxRmiServer, wrapper.getMBeanServer()); + + jmxConnectorServer.start(); + + registry.setRemoteServerStub(jmxRmiServer.toStub()); + JMXServerUtils.logJmxServiceUrl(addr, jmxPort); + waitForJmxAvailability(hostname, jmxPort, env); + } + catch (Throwable t) + { + 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 + { + 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) + { + 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); + } + throw new RuntimeException("Could not start JMX - unreachable after 20 attempts", lastThrown); + } + + public void stopJmx() throws IllegalAccessException, NoSuchFieldException, InterruptedException + { + if (!config.has(JMX)) + return; + // First, swap the mbean wrapper back to a NoOp wrapper + // This prevents later attempts to unregister mbeans from failing in Cassandra code, as we're going to + // unregister all of them here + ((MBeanWrapper.DelegatingMbeanWrapper) MBeanWrapper.instance).setDelegate(new MBeanWrapper.NoOpMBeanWrapper()); + try + { + wrapper.close(); + } + catch (Throwable e) + { + inInstancelogger.warn("failed to close wrapper.", e); + } + try + { + jmxConnectorServer.stop(); + } + catch (Throwable e) + { + inInstancelogger.warn("failed to close jmxConnectorServer.", e); + } + try + { + registry.close(); + } + catch (Throwable e) + { + inInstancelogger.warn("failed to close registry.", e); + } + try + { + serverSocketFactory.close(); + } + catch (Throwable e) + { + inInstancelogger.warn("failed to close serverSocketFactory.", e); + } + // 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 + } + + private void clearMapField(Class clazz, Object instance, String mapName) + throws IllegalAccessException, NoSuchFieldException { + Field mapField = ReflectionUtils.getField(clazz, 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(); + } + } + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/ResourceLeakTest.java b/test/distributed/org/apache/cassandra/distributed/test/ResourceLeakTest.java index a5a2dce171..794873feea 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/ResourceLeakTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/ResourceLeakTest.java @@ -26,21 +26,29 @@ import java.sql.Date; import java.text.SimpleDateFormat; import java.util.function.Consumer; import javax.management.MBeanServer; +import javax.management.MBeanServerConnection; +import javax.management.remote.JMXConnector; -import org.apache.cassandra.io.util.File; +import org.junit.Assert; import org.junit.Ignore; import org.junit.Test; import com.sun.management.HotSpotDiagnosticMXBean; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.ConsistencyLevel; +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.io.util.File; import org.apache.cassandra.utils.SigarLibrary; 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.utils.FBUtilities.now; +import static org.hamcrest.Matchers.startsWith; /* Resource Leak Test - useful when tracking down issues with in-JVM framework cleanup. * All objects referencing the InstanceClassLoader need to be garbage collected or @@ -138,6 +146,11 @@ public class ResourceLeakTest extends TestBaseImpl } void doTest(int numClusterNodes, Consumer updater) throws Throwable + { + doTest(numClusterNodes, updater, ignored -> {}); + } + + void doTest(int numClusterNodes, Consumer updater, Consumer actionToPerform) throws Throwable { for (int loop = 0; loop < numTestLoops; loop++) { @@ -149,6 +162,7 @@ public class ResourceLeakTest extends TestBaseImpl cluster.schemaChange("CREATE TABLE " + KEYSPACE + "." + tableName + " (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + "." + tableName + "(pk,ck,v) VALUES (0,0,0)", ConsistencyLevel.ALL); cluster.get(1).flush(KEYSPACE); + actionToPerform.accept(cluster); if (dumpEveryLoop) { dumpResources(String.format("loop%03d", loop)); @@ -207,4 +221,50 @@ public class ResourceLeakTest extends TestBaseImpl } dumpResources("final-native"); } + + @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); + } + } + }); + if (forceCollection) + { + System.runFinalization(); + System.gc(); + Thread.sleep(finalWaitMillis); + } + dumpResources("final-jmx"); + } + + @Test + public void looperEverythingTest() throws Throwable + { + doTest(1, config -> config.with(Feature.values())); + if (forceCollection) + { + System.runFinalization(); + System.gc(); + Thread.sleep(finalWaitMillis); + } + dumpResources("final-everything"); + } } diff --git a/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXFeatureTest.java b/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXFeatureTest.java new file mode 100644 index 0000000000..83a35e594d --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXFeatureTest.java @@ -0,0 +1,113 @@ +/* + * 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.test.jmx; + +import java.io.IOException; +import java.util.HashSet; +import java.util.Set; +import javax.management.MBeanServerConnection; +import javax.management.remote.JMXConnector; + +import org.junit.Assert; +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.impl.INodeProvisionStrategy; +import org.apache.cassandra.distributed.shared.JMXUtil; +import org.apache.cassandra.distributed.test.TestBaseImpl; + +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 + * - Test that when connecting, we get the correct MBeanServer by checking the default domain, which is set to the IP of the instance + * - Run the test multiple times to ensure cleanup of the JMX servers is complete so the next test can run successfully using the same host/port. + * + * @throws Exception + */ + @Test + public void testMultipleNetworkInterfacesProvisioning() throws Exception + { + int iterations = 2; // Make sure the JMX infrastructure all cleans up properly by running this multiple times. + Set allInstances = new HashSet<>(); + for (int i = 0; i < iterations; i++) + { + try (Cluster cluster = Cluster.build(2) + .withNodeProvisionStrategy(INodeProvisionStrategy.Strategy.MultipleNetworkInterfaces) + .withConfig(c -> c.with(Feature.values())).start()) + { + Set instancesContacted = new HashSet<>(); + for (IInvokableInstance instance : cluster.get(1, 2)) + { + testInstance(instancesContacted, instance); + } + Assert.assertEquals("Should have connected with both JMX instances.", 2, instancesContacted.size()); + allInstances.addAll(instancesContacted); + } + } + Assert.assertEquals("Each instance from each cluster should have been unique", iterations * 2, allInstances.size()); + } + + @Test + public void testOneNetworkInterfaceProvisioning() throws Exception + { + int iterations = 2; // Make sure the JMX infrastructure all cleans up properly by running this multiple times. + Set allInstances = new HashSet<>(); + for (int i = 0; i < iterations; i++) + { + try (Cluster cluster = Cluster.build(2) + .withNodeProvisionStrategy(INodeProvisionStrategy.Strategy.OneNetworkInterface) + .withConfig(c -> c.with(Feature.values())).start()) + { + Set instancesContacted = new HashSet<>(); + for (IInvokableInstance instance : cluster.get(1, 2)) + { + testInstance(instancesContacted, instance); + } + Assert.assertEquals("Should have connected with both JMX instances.", 2, instancesContacted.size()); + allInstances.addAll(instancesContacted); + } + } + Assert.assertEquals("Each instance from each cluster should have been unique", iterations * 2, allInstances.size()); + } + + private void testInstance(Set instancesContacted, IInvokableInstance instance) throws IOException + { + // 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. + 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(); + instancesContacted.add(defaultDomain); + Assert.assertThat(defaultDomain, startsWith(JMXUtil.getJmxHost(config) + ":" + config.jmxPort())); + } + } +} 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 ef0f1ecae0..ece6052a75 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXGetterCheckTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXGetterCheckTest.java @@ -17,7 +17,6 @@ */ package org.apache.cassandra.distributed.test.jmx; -import java.net.InetAddress; import java.util.ArrayList; import java.util.Collections; import java.util.List; @@ -31,7 +30,6 @@ import javax.management.MBeanServerConnection; import javax.management.ObjectName; import javax.management.remote.JMXConnector; import javax.management.remote.JMXConnectorFactory; -import javax.management.remote.JMXConnectorServer; import javax.management.remote.JMXServiceURL; import com.google.common.collect.ImmutableSet; @@ -39,11 +37,8 @@ import org.junit.Test; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.distributed.api.IInvokableInstance; import org.apache.cassandra.distributed.test.TestBaseImpl; -import org.apache.cassandra.utils.JMXServerUtils; - -import static org.apache.cassandra.config.CassandraRelevantProperties.IS_DISABLED_MBEAN_REGISTRATION; -import static org.apache.cassandra.cql3.CQLTester.getAutomaticallyAllocatedPort; public class JMXGetterCheckTest extends TestBaseImpl { @@ -57,20 +52,17 @@ public class JMXGetterCheckTest extends TestBaseImpl "org.apache.cassandra.db:type=StorageService:resetLocalSchema" // this will fail when there are no other nodes which can serve schema ); - @Test - public void test() throws Exception - { - // start JMX server, which the instance will register with - InetAddress loopback = InetAddress.getLoopbackAddress(); - String jmxHost = loopback.getHostAddress(); - int jmxPort = getAutomaticallyAllocatedPort(loopback); - JMXConnectorServer jmxServer = JMXServerUtils.createJMXServer(jmxPort, true); - jmxServer.start(); - String url = "service:jmx:rmi:///jndi/rmi://" + jmxHost + ":" + jmxPort + "/jmxrmi"; + public static final String JMX_SERVICE_URL_FMT = "service:jmx:rmi:///jndi/rmi://%s:%d/jmxrmi"; - IS_DISABLED_MBEAN_REGISTRATION.setBoolean(false); + @Test + public void testGetters() throws Exception + { try (Cluster cluster = Cluster.build(1).withConfig(c -> c.with(Feature.values())).start()) { + IInvokableInstance instance = cluster.get(1); + + String jmxHost = instance.config().broadcastAddress().getAddress().getHostAddress(); + String url = String.format(JMX_SERVICE_URL_FMT, jmxHost, instance.config().jmxPort()); List errors = new ArrayList<>(); try (JMXConnector jmxc = JMXConnectorFactory.connect(new JMXServiceURL(url), null)) { @@ -122,7 +114,7 @@ public class JMXGetterCheckTest extends TestBaseImpl } /** - * This class is meant to make new errors easier to read, by adding the JMX endpoint, and cleaning up the unneded JMX/Reflection logic cluttering the stacktrace + * This class is meant to make new errors easier to read, by adding the JMX endpoint, and cleaning up the unneeded JMX/Reflection logic cluttering the stacktrace */ private static class Named extends RuntimeException { diff --git a/test/distributed/org/apache/cassandra/distributed/test/metric/TableMetricTest.java b/test/distributed/org/apache/cassandra/distributed/test/metric/TableMetricTest.java index 08c9324649..6725494952 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/metric/TableMetricTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/metric/TableMetricTest.java @@ -29,7 +29,9 @@ import java.util.concurrent.ConcurrentMap; import java.util.stream.Collectors; import javax.management.InstanceAlreadyExistsException; import javax.management.InstanceNotFoundException; +import javax.management.MBeanServer; import javax.management.ObjectName; +import javax.management.QueryExp; import com.google.common.collect.ImmutableSet; import org.junit.Assert; @@ -57,7 +59,6 @@ public class TableMetricTest extends TestBaseImpl MBEAN_REGISTRATION_CLASS.setString(MapMBeanWrapper.class.getName()); IS_DISABLED_MBEAN_REGISTRATION.setBoolean(false); } - private static volatile Map> SYSTEM_TABLES = null; private static Set TABLE_METRIC_NAMES = ImmutableSet.of("WriteLatency"); @@ -165,7 +166,7 @@ public class TableMetricTest extends TestBaseImpl { inst.runOnInstance(() -> { // cast only to make sure it linked properly - MapMBeanWrapper mbeans = (MapMBeanWrapper) MBeanWrapper.instance; + MapMBeanWrapper mbeans = getMapMBeanWrapper(); Assert.assertTrue("Unable to find table mbean for " + keyspace + "." + table, mbeans.isRegistered(ColumnFamilyStore.getTableMBeanName(keyspace, table, false))); Assert.assertTrue("Unable to find column family mbean for " + keyspace + "." + table, @@ -177,7 +178,7 @@ public class TableMetricTest extends TestBaseImpl { inst.runOnInstance(() -> { // cast only to make sure it linked properly - MapMBeanWrapper mbeans = (MapMBeanWrapper) MBeanWrapper.instance; + MapMBeanWrapper mbeans = getMapMBeanWrapper(); Assert.assertFalse("Found table mbean for " + keyspace + "." + table, mbeans.isRegistered(ColumnFamilyStore.getTableMBeanName(keyspace, table, false))); Assert.assertFalse("Found column family mbean for " + keyspace + "." + table, @@ -189,7 +190,7 @@ public class TableMetricTest extends TestBaseImpl { inst.runOnInstance(() -> { // cast only to make sure it linked properly - MapMBeanWrapper mbeans = (MapMBeanWrapper) MBeanWrapper.instance; + MapMBeanWrapper mbeans = getMapMBeanWrapper(); String mbean = getTableMetricName(keyspace, table, name); Assert.assertTrue("Unable to find metric " + name + " for " + keyspace + "." + table, mbeans.isRegistered(mbean)); @@ -203,7 +204,7 @@ public class TableMetricTest extends TestBaseImpl { inst.runOnInstance(() -> { // cast only to make sure it linked properly - MapMBeanWrapper mbeans = (MapMBeanWrapper) MBeanWrapper.instance; + MapMBeanWrapper mbeans = getMapMBeanWrapper(); String mbean = getTableMetricName(keyspace, table, name); Assert.assertFalse("Found metric " + name + " for " + keyspace + "." + table, mbeans.isRegistered(mbean)); @@ -226,7 +227,7 @@ public class TableMetricTest extends TestBaseImpl { inst.runOnInstance(() -> { // cast only to make sure it linked properly - MapMBeanWrapper mbeans = (MapMBeanWrapper) MBeanWrapper.instance; + MapMBeanWrapper mbeans = getMapMBeanWrapper(); String keyspaceMBean = getKeyspaceMetricName(keyspace, name); Assert.assertFalse("Found keyspace metric " + keyspaceMBean + " for " + keyspace, mbeans.isRegistered(keyspaceMBean)); @@ -243,6 +244,10 @@ public class TableMetricTest extends TestBaseImpl return String.format("org.apache.cassandra.metrics:type=Table,keyspace=%s,scope=%s,name=%s", keyspace, table, name); } + private static MapMBeanWrapper getMapMBeanWrapper() + { + return (MapMBeanWrapper) ((MBeanWrapper.DelegatingMbeanWrapper)MBeanWrapper.instance).getDelegate(); + } public static final class MapMBeanWrapper implements MBeanWrapper { private final ConcurrentMap map = new ConcurrentHashMap<>(); @@ -268,5 +273,17 @@ public class TableMetricTest extends TestBaseImpl if (previous == null) onException.handler.accept(new InstanceNotFoundException("MBean " + mbeanName + " was not found")); } + + @Override + public Set queryNames(ObjectName name, QueryExp query) + { + return null; + } + + @Override + public MBeanServer getMBeanServer() + { + return null; + } } }