natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-4.1' into trunk

This commit is contained in:
Jon Meredith 2023-05-31 12:27:36 -06:00
parent cccc46cb23 208a71c008
commit 4bcadc6bfb
20 changed files with 922 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.build/parent-pom-template.xml
View File

@ -489,7 +489,7 @@
<dependency>
<groupId>org.apache.cassandra</groupId>
<artifactId>dtest-api</artifactId>
<version>0.0.13</version>
<version>0.0.15</version>
<scope>test</scope>
</dependency>
<dependency>

27
build.xml
View File

@ -207,6 +207,7 @@
<string>--add-exports java.management.rmi/com.sun.jmx.remote.internal.rmi=ALL-UNNAMED</string>
<string>--add-exports java.rmi/sun.rmi.registry=ALL-UNNAMED</string>
<string>--add-exports java.rmi/sun.rmi.server=ALL-UNNAMED</string>
<string>--add-exports java.rmi/sun.rmi.transport.tcp=ALL-UNNAMED</string>
<string>--add-exports java.sql/java.sql=ALL-UNNAMED</string>
<string>--add-opens java.base/java.lang.module=ALL-UNNAMED</string>
@ -237,6 +238,7 @@
<string>--add-exports java.management.rmi/com.sun.jmx.remote.internal.rmi=ALL-UNNAMED</string>
<string>--add-exports java.rmi/sun.rmi.registry=ALL-UNNAMED</string>
<string>--add-exports java.rmi/sun.rmi.server=ALL-UNNAMED</string>
<string>--add-exports java.rmi/sun.rmi.transport.tcp=ALL-UNNAMED</string>
<string>--add-exports java.sql/java.sql=ALL-UNNAMED</string>
<string>--add-exports java.base/java.lang.ref=ALL-UNNAMED</string>
<string>--add-exports jdk.unsupported/sun.misc=ALL-UNNAMED</string>
@ -263,6 +265,8 @@
<string>--add-opens java.base/java.math=ALL-UNNAMED</string>
<string>--add-opens java.base/java.lang.reflect=ALL-UNNAMED</string>
<string>--add-opens java.base/java.net=ALL-UNNAMED</string>
<string>--add-opens java.rmi/sun.rmi.transport.tcp=ALL-UNNAMED</string>
</resources>
<pathconvert property="_jvm_args_concat2" refid="_jvm17_arg_items" pathsep=" "/>
<condition property="java17-jvmargs" value="${_jvm_args_concat2}" else="">
@ -307,7 +311,8 @@
</condition>
<!-- needed to compile org.apache.cassandra.utils.JMXServerUtils -->
<condition property="jdk11plus-javac-exports" value="--add-exports java.rmi/sun.rmi.registry=ALL-UNNAMED" else="">
<!-- needed to compile org.apache.cassandra.distributed.impl.Instance-->
<condition property="jdk11plus-javac-exports" value="--add-exports java.rmi/sun.rmi.registry=ALL-UNNAMED --add-exports java.rmi/sun.rmi.transport.tcp=ALL-UNNAMED" else="">
<not>
<equals arg1="${ant.java.version}" arg2="1.8"/>
</not>
@ -1035,6 +1040,7 @@
<src path="${test.simulator-asm.src}"/>
<src path="${test.simulator-bootstrap.src}"/>
<src path="${test.simulator-test.src}"/>
<compilerarg line="${jdk11plus-javac-exports}"/>
</javac>
<!-- Non-java resources needed by the test suite -->
@ -1811,6 +1817,24 @@
<echo>"IDE configuration updated for use with JDK11"</echo>
</target>
<target name="_maybe_update_idea_to_java17" depends="init" if="java.version.17">
<replace file="${eclipse.project.name}.iml" token="JDK_1_8" value="JDK_17"/>
<replace file=".idea/misc.xml" token="JDK_1_8" value="JDK_17"/>
<replace file=".idea/misc.xml" token="1.8" value="17"/>
<replaceregexp file=".idea/workspace.xml"
match="name=&quot;VM_PARAMETERS&quot; value=&quot;(.*)"
replace="name=&quot;VM_PARAMETERS&quot; value=&quot;\1 ${java17-jvmargs}"
byline="true"/>
<echo file=".idea/compiler.xml"><![CDATA[<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="JavacSettings">
<option name="ADDITIONAL_OPTIONS_STRING" value="--add-exports java.rmi/sun.rmi.registry=ALL-UNNAMED --add-exports java.rmi/sun.rmi.transport.tcp=ALL-UNNAMED" />
</component>
</project>]]></echo>
<echo>"IDE configuration updated for use with JDK17"</echo>
</target>
<!-- Generate IDEA project description files -->
<target name="generate-idea-files" depends="init,resolver-dist-lib,gen-cql3-grammar,generate-jflex-java,createVersionPropFile" description="Generate IDEA files">
<mkdir dir=".idea"/>
@ -1831,6 +1855,7 @@
</component>
</project>]]></echo>
<antcall target="_maybe_update_idea_to_java11"/>
<antcall target="_maybe_update_idea_to_java17"/>
</target>
<!-- Generate Eclipse project description files -->

12
src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
View File

@ -200,6 +200,8 @@ public enum CassandraRelevantProperties
DRAIN_EXECUTOR_TIMEOUT_MS("cassandra.drain_executor_timeout_ms", convertToString(TimeUnit.MINUTES.toMillis(5))),
DROP_OVERSIZED_READ_REPAIR_MUTATIONS("cassandra.drop_oversized_readrepair_mutations"),
DTEST_API_LOG_TOPOLOGY("cassandra.dtest.api.log.topology"),
/** 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"),
ENABLE_DC_LOCAL_COMMIT("cassandra.enable_dc_local_commit", "true"),
/**
* Whether {@link org.apache.cassandra.db.ConsistencyLevel#NODE_LOCAL} should be allowed.
@ -262,6 +264,11 @@ public enum CassandraRelevantProperties
* Default is absolute path to lib directory.
*/
JAVA_LIBRARY_PATH("java.library.path"),
/**
* 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 value of this property represents the host name string
* that should be associated with remote stubs for locally created remote objects,
@ -421,6 +428,11 @@ public enum CassandraRelevantProperties
/** Platform word size sun.arch.data.model. Examples: "32", "64", "unknown"*/
SUN_ARCH_DATA_MODEL("sun.arch.data.model"),
SUN_JAVA_COMMAND("sun.java.command", ""),
/**
* 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"),
SUN_STDERR_ENCODING("sun.stderr.encoding"),
SUN_STDOUT_ENCODING("sun.stdout.encoding"),
SUPERUSER_SETUP_DELAY_MS("cassandra.superuser_setup_delay_ms", "10000"),

8
src/java/org/apache/cassandra/service/GCInspector.java
View File

@ -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)

3
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -987,7 +987,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
completeInitialization();
}
private void completeInitialization()
@VisibleForTesting
public void completeInitialization()
{
if (!initialized)
registerMBeans();

28
src/java/org/apache/cassandra/utils/JMXServerUtils.java
View File

@ -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;
@ -257,7 +259,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;
@ -330,11 +333,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 {
@ -367,5 +370,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
}
}
}
}

208
src/java/org/apache/cassandra/utils/MBeanWrapper.java
View File

@ -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.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 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<ObjectName> 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<ObjectName> 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<ObjectName> 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<ObjectName> 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<ObjectName> 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); }),

62
src/java/org/apache/cassandra/utils/RMIClientSocketFactoryImpl.java Normal file
View File

@ -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);
}
}

11
src/java/org/apache/cassandra/utils/ReflectionUtils.java
View File

@ -29,11 +29,16 @@ public class ReflectionUtils
}
public static Field getModifiersField() throws NoSuchFieldException
{
return getField(Field.class, "modifiers");
}
public static Field getField(Class<?> clazz, String fieldName) throws NoSuchFieldException
{
// below code works before Java 12
try
{
return Field.class.getDeclaredField("modifiers");
return clazz.getDeclaredField(fieldName);
}
catch (NoSuchFieldException e)
{
@ -42,10 +47,10 @@ public class ReflectionUtils
{
Method getDeclaredFields0 = Class.class.getDeclaredMethod("getDeclaredFields0", boolean.class);
getDeclaredFields0.setAccessible(true);
Field[] fields = (Field[]) getDeclaredFields0.invoke(Field.class, false);
Field[] fields = (Field[]) getDeclaredFields0.invoke(clazz, false);
for (Field field : fields)
{
if ("modifiers".equals(field.getName()))
if (fieldName.equals(field.getName()))
{
return field;
}

2
test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
View File

@ -186,6 +186,8 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
private ShutdownExecutor shutdownExecutor = DEFAULT_SHUTDOWN_EXECUTOR;
{
// Indicate that we are running in the in-jvm dtest environment
CassandraRelevantProperties.DTEST_IS_IN_JVM_DTEST.setBoolean(true);
// those properties may be set for unit-test optimizations; those should not be used when running dtests
CassandraRelevantProperties.TEST_FLUSH_LOCAL_SCHEMA_CHANGES.reset();
CassandraRelevantProperties.NON_GRACEFUL_SHUTDOWN.reset();

88
test/distributed/org/apache/cassandra/distributed/impl/CollectingRMIServerSocketFactoryImpl.java Normal file
View File

@ -0,0 +1,88 @@
/*
* 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<ServerSocket> 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);
}
}

11
test/distributed/org/apache/cassandra/distributed/impl/INodeProvisionStrategy.java
View File

@ -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);
}

22
test/distributed/org/apache/cassandra/distributed/impl/Instance.java
View File

@ -160,6 +160,7 @@ import static org.apache.cassandra.config.CassandraRelevantProperties.TEST_CASSA
import static org.apache.cassandra.config.CassandraRelevantProperties.TEST_CASSANDRA_TESTTAG;
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;
@ -177,6 +178,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
private volatile boolean initialized = false;
private volatile boolean internodeMessagingStarted = false;
private final AtomicLong startedAt = new AtomicLong();
private IsolatedJmx isolatedJmx;
@Deprecated
Instance(IInstanceConfig config, ClassLoader classLoader)
@ -603,6 +605,9 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
config.networkTopology(), config.broadcastAddress());
DistributedTestSnitch.assign(config.networkTopology());
if (config.has(JMX))
startJmx();
DatabaseDescriptor.daemonInitialization();
LoggingSupportFactory.getLoggingSupport().onStartup();
@ -723,6 +728,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
StorageService.instance.setNormalModeUnsafe();
Gossiper.instance.register(StorageService.instance);
StorageService.instance.startSnapshotManager();
StorageService.instance.completeInitialization();
}
// Populate tokenMetadata for the second time,
@ -759,6 +765,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)
@ -900,6 +920,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

16
test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
View File

@ -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<String> 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)

230
test/distributed/org/apache/cassandra/distributed/impl/IsolatedJmx.java Normal file
View File

@ -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<String, Object> 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<String, Object> 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 <K, V> void clearMapField(Class<?> clazz, Object instance, String mapName)
throws IllegalAccessException, NoSuchFieldException {
Field mapField = ReflectionUtils.getField(clazz, mapName);
mapField.setAccessible(true);
Map<K, V> map = (Map<K, V>) mapField.get(instance);
// Because multiple instances can be shutting down at once,
// synchronize on the map to avoid ConcurrentModificationException
synchronized (map)
{
for (Iterator<Map.Entry<K, V>> it = map.entrySet().iterator(); it.hasNext(); )
{
it.next();
it.remove();
}
}
}
}

62
test/distributed/org/apache/cassandra/distributed/test/ResourceLeakTest.java
View File

@ -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<IInstanceConfig> updater) throws Throwable
{
doTest(numClusterNodes, updater, ignored -> {});
}
void doTest(int numClusterNodes, Consumer<IInstanceConfig> updater, Consumer<Cluster> 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");
}
}

14
test/distributed/org/apache/cassandra/distributed/test/VirtualTableLogsTest.java
View File

@ -27,6 +27,7 @@ import ch.qos.logback.classic.Level;
import org.apache.cassandra.db.virtual.LogMessagesTable;
import org.apache.cassandra.db.virtual.LogMessagesTable.LogMessage;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.Constants;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.SimpleQueryResult;
import org.apache.cassandra.distributed.shared.WithProperties;
@ -68,12 +69,17 @@ public class VirtualTableLogsTest extends TestBaseImpl
public void testMultipleAppendersFailToStartNode() throws Throwable
{
LOGBACK_CONFIGURATION_FILE.setString("test/conf/logback-dtest_with_vtable_appender_invalid.xml");
// NOTE: Because cluster startup is expected to fail in this case, and can leave things in a weird state
// for the next state, create without starting, and set failure as shutdown to false,
// so the try-with-resources can close instances properly.
try (WithProperties properties = new WithProperties().set(LOGBACK_CONFIGURATION_FILE, "test/conf/logback-dtest_with_vtable_appender_invalid.xml");
Cluster ignored = Cluster.build(1)
.withConfig(c -> c.with(Feature.values()))
.start();
)
Cluster cluster = Cluster.build(1)
.withConfig(c -> c.with(Feature.values())
.set(Constants.KEY_DTEST_API_STARTUP_FAILURE_AS_SHUTDOWN, false))
.createWithoutStarting())
{
cluster.startup();
fail("Node should not start as there is supposed to be invalid logback configuration file.");
}
catch (IllegalStateException ex)

113
test/distributed/org/apache/cassandra/distributed/test/jmx/JMXFeatureTest.java Normal file
View File

@ -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<String> 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<String> 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<String> 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<String> 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<String> 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()));
}
}
}

28
test/distributed/org/apache/cassandra/distributed/test/jmx/JMXGetterCheckTest.java
View File

@ -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.ORG_APACHE_CASSANDRA_DISABLE_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";
ORG_APACHE_CASSANDRA_DISABLE_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<Named> 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
{

29
test/distributed/org/apache/cassandra/distributed/test/metric/TableMetricTest.java
View File

@ -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());
ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION.setBoolean(false);
}
private static volatile Map<String, Collection<String>> SYSTEM_TABLES = null;
private static Set<String> 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<ObjectName, Object> 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<ObjectName> queryNames(ObjectName name, QueryExp query)
{
return null;
}
@Override
public MBeanServer getMBeanServer()
{
return null;
}
}
}