You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by jo...@apache.org on 2023/05/31 19:23:34 UTC
[cassandra] branch cassandra-4.0 updated (f66adb0278 -> 57293e8281)
This is an automated email from the ASF dual-hosted git repository.
jonmeredith pushed a change to branch cassandra-4.0
in repository https://gitbox.apache.org/repos/asf/cassandra.git
from f66adb0278 Remove unnecessary shuffling of GossipDigests in Gossiper#makeRandomGossipDigest
new 43ec184391 Add support for JMX in the in-jvm dtest framework
new 57293e8281 Merge branch 'cassandra-3.11' into cassandra-4.0
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.build/build-resolver.xml | 2 +
build.xml | 9 +-
relocate-dependencies.pom | 2 +-
.../config/CassandraRelevantProperties.java | 16 ++
.../org/apache/cassandra/service/GCInspector.java | 8 +-
.../java/org/apache/cassandra/utils}/Isolated.java | 4 +-
.../org/apache/cassandra/utils/JMXServerUtils.java | 28 ++-
.../org/apache/cassandra/utils/MBeanWrapper.java | 212 ++++++++++++++++++-
.../org/apache/cassandra/utils/NativeLibrary.java | 2 +
.../cassandra/utils/NativeLibraryDarwin.java | 1 +
.../apache/cassandra/utils/NativeLibraryLinux.java | 1 +
.../cassandra/utils/NativeLibraryWindows.java | 1 +
.../cassandra/utils/NativeLibraryWrapper.java | 1 +
.../utils/RMIClientSocketFactoryImpl.java | 62 ++++++
.../apache/cassandra/utils/ReflectionUtils.java | 55 +++++
.../java/org/apache/cassandra/utils}/Shared.java | 2 +-
.../distributed/impl/AbstractCluster.java | 16 +-
.../impl/CollectingRMIServerSocketFactoryImpl.java | 87 ++++++++
.../distributed/impl/INodeProvisionStrategy.java | 11 +
.../cassandra/distributed/impl/Instance.java | 33 +++
.../cassandra/distributed/impl/InstanceConfig.java | 18 +-
.../distributed/impl/IsolatedExecutor.java | 2 +-
.../cassandra/distributed/impl/IsolatedJmx.java | 230 +++++++++++++++++++++
.../cassandra/distributed/shared/ClusterUtils.java | 1 +
.../test/BootstrapBinaryDisabledTest.java | 2 +-
.../distributed/test/ResourceLeakTest.java | 60 ++++++
.../distributed/test/jmx/JMXFeatureTest.java | 113 ++++++++++
.../distributed/test/jmx/JMXGetterCheckTest.java | 136 ++++++++++++
.../distributed/test/metric/TableMetricTest.java | 29 ++-
.../upgrade/MixedModeMessageForwardTest.java | 2 +-
30 files changed, 1107 insertions(+), 39 deletions(-)
rename {test/distributed/org/apache/cassandra/distributed/shared => src/java/org/apache/cassandra/utils}/Isolated.java (95%)
create mode 100644 src/java/org/apache/cassandra/utils/RMIClientSocketFactoryImpl.java
create mode 100644 src/java/org/apache/cassandra/utils/ReflectionUtils.java
rename {test/distributed/org/apache/cassandra/distributed/shared => src/java/org/apache/cassandra/utils}/Shared.java (96%)
create mode 100644 test/distributed/org/apache/cassandra/distributed/impl/CollectingRMIServerSocketFactoryImpl.java
create mode 100644 test/distributed/org/apache/cassandra/distributed/impl/IsolatedJmx.java
create mode 100644 test/distributed/org/apache/cassandra/distributed/test/jmx/JMXFeatureTest.java
create mode 100644 test/distributed/org/apache/cassandra/distributed/test/jmx/JMXGetterCheckTest.java
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
[cassandra] 01/01: Merge branch 'cassandra-3.11' into cassandra-4.0
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jonmeredith pushed a commit to branch cassandra-4.0
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 57293e8281774f26ad4092b97ae6e17eb3f5a7e9
Merge: f66adb0278 43ec184391
Author: Jon Meredith <jo...@apache.org>
AuthorDate: Wed May 31 12:02:29 2023 -0600
Merge branch 'cassandra-3.11' into cassandra-4.0
.build/build-resolver.xml | 2 +
build.xml | 9 +-
relocate-dependencies.pom | 2 +-
.../config/CassandraRelevantProperties.java | 16 ++
.../org/apache/cassandra/service/GCInspector.java | 8 +-
.../java/org/apache/cassandra/utils}/Isolated.java | 4 +-
.../org/apache/cassandra/utils/JMXServerUtils.java | 28 ++-
.../org/apache/cassandra/utils/MBeanWrapper.java | 212 ++++++++++++++++++-
.../org/apache/cassandra/utils/NativeLibrary.java | 2 +
.../cassandra/utils/NativeLibraryDarwin.java | 1 +
.../apache/cassandra/utils/NativeLibraryLinux.java | 1 +
.../cassandra/utils/NativeLibraryWindows.java | 1 +
.../cassandra/utils/NativeLibraryWrapper.java | 1 +
.../utils/RMIClientSocketFactoryImpl.java | 62 ++++++
.../apache/cassandra/utils/ReflectionUtils.java | 55 +++++
.../java/org/apache/cassandra/utils}/Shared.java | 2 +-
.../distributed/impl/AbstractCluster.java | 16 +-
.../impl/CollectingRMIServerSocketFactoryImpl.java | 87 ++++++++
.../distributed/impl/INodeProvisionStrategy.java | 11 +
.../cassandra/distributed/impl/Instance.java | 33 +++
.../cassandra/distributed/impl/InstanceConfig.java | 18 +-
.../distributed/impl/IsolatedExecutor.java | 2 +-
.../cassandra/distributed/impl/IsolatedJmx.java | 230 +++++++++++++++++++++
.../cassandra/distributed/shared/ClusterUtils.java | 1 +
.../test/BootstrapBinaryDisabledTest.java | 2 +-
.../distributed/test/ResourceLeakTest.java | 60 ++++++
.../distributed/test/jmx/JMXFeatureTest.java | 113 ++++++++++
.../distributed/test/jmx/JMXGetterCheckTest.java | 136 ++++++++++++
.../distributed/test/metric/TableMetricTest.java | 29 ++-
.../upgrade/MixedModeMessageForwardTest.java | 2 +-
30 files changed, 1107 insertions(+), 39 deletions(-)
diff --cc .build/build-resolver.xml
index 03db1c8671,99bfe4bdc7..572ca996b7
--- a/.build/build-resolver.xml
+++ b/.build/build-resolver.xml
@@@ -56,42 -51,12 +56,44 @@@
<resolver:remoterepos id="all">
<remoterepo id="resolver-central" url="${artifact.remoteRepository.central}"/>
<remoterepo id="resolver-apache" url="${artifact.remoteRepository.apache}"/>
- <!-- Only needed for PR builds - remove before commit -->
- <!-- <remoterepo id="resolver-apache-snapshot" url="https://repository.apache.org/content/repositories/snapshots" releases="false" snapshots="true" updates="always" checksums="fail" />-->
++ <!-- Snapshots are not allowed, but for feature branches they may be needed, so uncomment the below to allow snapshots to work -->
++ <!-- <remoterepo id="resolver-apache-snapshot" url="https://repository.apache.org/content/repositories/snapshots" releases="false" snapshots="true" updates="always" checksums="fail" /> -->
</resolver:remoterepos>
- <resolver:resolve>
- <remoterepos refid="all"/>
+ <macrodef name="resolve">
+ <!--
+ maven-resolver-ant-tasks's resolve logic doesn't have retry logic and does not respect settings.xml,
+ this causes issues when overriding maven central is required (such as when behind a corporate firewall);
+ it is critical to always provide the 'all' remoterepos to override resolve's default hard coded logic.
+
+ This macro simplifies this logic by centralizing retries and forcing the remoterepo 'all' to be used.
+ -->
+ <attribute name="failonmissingattachments" default="true"/>
+ <element name="elements" implicit="yes"/>
+ <sequential>
+ <retry retrycount="3">
+ <resolver:resolve failonmissingattachments="@{failonmissingattachments}">
+ <resolver:remoterepos refid="all"/>
+ <elements/>
+ </resolver:resolve>
+ </retry>
+ </sequential>
+ </macrodef>
+ <macrodef name="resolvepom">
+ <attribute name="file"/>
+ <attribute name="id"/>
+ <element name="elements" implicit="yes" optional="yes"/>
+ <sequential>
+ <retry retrycount="3">
+ <resolver:pom file="@{file}" id="@{id}">
+ <remoterepos refid="all"/>
+ <elements/>
+ </resolver:pom>
+ </retry>
+ </sequential>
+ </macrodef>
+
+ <resolve>
<dependencies>
<dependency groupId="com.datastax.wikitext" artifactId="wikitext-core-ant" version="1.3"/>
<dependency groupId="org.fusesource.wikitext" artifactId="textile-core" version="1.3"/>
diff --cc build.xml
index f9dc014f26,8f26c5e8f2..f50cb9bf4d
--- a/build.xml
+++ b/build.xml
@@@ -176,109 -160,6 +176,111 @@@
}
</script>
+ <condition property="java.version.8">
+ <equals arg1="${ant.java.version}" arg2="1.8"/>
+ </condition>
+ <condition property="java.version.11">
+ <not><isset property="java.version.8"/></not>
+ </condition>
+ <fail><condition><not><or>
+ <isset property="java.version.8"/>
+ <isset property="java.version.11"/>
+ </or></not></condition></fail>
+
+ <resources id="_jvm11_arg_items">
+ <string>-Djdk.attach.allowAttachSelf=true</string>
+
+ <string>-XX:+UseConcMarkSweepGC</string>
+ <string>-XX:+CMSParallelRemarkEnabled</string>
+ <string>-XX:SurvivorRatio=8</string>
+ <string>-XX:MaxTenuringThreshold=1</string>
+ <string>-XX:CMSInitiatingOccupancyFraction=75</string>
+ <string>-XX:+UseCMSInitiatingOccupancyOnly</string>
+ <string>-XX:CMSWaitDuration=10000</string>
+ <string>-XX:+CMSParallelInitialMarkEnabled</string>
+ <string>-XX:+CMSEdenChunksRecordAlways</string>
+
+ <string>--add-exports java.base/jdk.internal.misc=ALL-UNNAMED</string>
+ <string>--add-exports java.base/jdk.internal.ref=ALL-UNNAMED</string>
+ <string>--add-exports java.base/sun.nio.ch=ALL-UNNAMED</string>
+ <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>
+ <string>--add-opens java.base/java.net=ALL-UNNAMED</string>
+ <string>--add-opens java.base/jdk.internal.loader=ALL-UNNAMED</string>
+ <string>--add-opens java.base/jdk.internal.ref=ALL-UNNAMED</string>
+ <string>--add-opens java.base/jdk.internal.reflect=ALL-UNNAMED</string>
+ <string>--add-opens java.base/jdk.internal.math=ALL-UNNAMED</string>
+ <string>--add-opens java.base/jdk.internal.module=ALL-UNNAMED</string>
+ <string>--add-opens java.base/jdk.internal.util.jar=ALL-UNNAMED</string>
+ <string>--add-opens jdk.management/com.sun.management.internal=ALL-UNNAMED</string>
++
+ </resources>
+ <pathconvert property="_jvm_args_concat" refid="_jvm11_arg_items" pathsep=" "/>
+ <condition property="java11-jvmargs" value="${_jvm_args_concat}" else="">
+ <not>
+ <equals arg1="${ant.java.version}" arg2="1.8"/>
+ </not>
+ </condition>
+
+ <!--
+ JVM arguments for tests.
+
+ There is a race condition bug in java 11 (see CASSANDRA-15981) which causes a crash of the
+ JVM; this race is between CMS and class unloading. In java 8 we can cap the metaspace to
+ make tests stable on low resource environments, but in java 11 we need to make it unlimited
+ (don't define MaxMetaspaceSize) and disable class unloading in CMS outside of a
+ stop-the-world pause.
+
+ In java 11 we also need to set a system property to enable netty to use Unsafe direct byte
+ buffer construction (see CASSANDRA-16493)
+ -->
+ <resources id="_jvm8_test_arg_items">
+ <!-- TODO see CASSANDRA-16212 - we seem to OOM non stop now after CASSANDRA-16212, so to have clean CI while this gets looked into, disabling limiting metaspace
+ <string>-XX:MaxMetaspaceExpansion=64M</string>
+ <string>-XX:MaxMetaspaceSize=512M</string>
+ <string>-XX:MetaspaceSize=128M</string>
+ -->
+ </resources>
+ <pathconvert property="_jvm8_test_arg_items_concat" refid="_jvm8_test_arg_items" pathsep=" "/>
+ <resources id="_jvm11_test_arg_items">
+ <string>-XX:-CMSClassUnloadingEnabled</string>
+ <string>-Dio.netty.tryReflectionSetAccessible=true</string>
+ </resources>
+ <pathconvert property="_jvm11_test_arg_items_concat" refid="_jvm11_test_arg_items" pathsep=" "/>
+ <condition property="test-jvmargs" value="${_jvm11_test_arg_items_concat}" else="${_jvm8_test_arg_items_concat}">
+ <not>
+ <equals arg1="${ant.java.version}" arg2="1.8"/>
+ </not>
+ </condition>
+
+ <!-- needed to compile org.apache.cassandra.utils.JMXServerUtils -->
- <condition property="jdk11-javac-exports" value="--add-exports java.rmi/sun.rmi.registry=ALL-UNNAMED" else="">
++ <condition property="jdk11-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>
+ </condition>
+ <condition property="jdk11-javadoc-exports" value="${jdk11-javac-exports} --frames" else="">
+ <not>
+ <equals arg1="${ant.java.version}" arg2="1.8"/>
+ </not>
+ </condition>
+
+ <condition property="build.java.11">
+ <istrue value="${use.jdk11}"/>
+ </condition>
+
+ <condition property="source.version" value="8" else="11">
+ <equals arg1="${java.version.8}" arg2="true"/>
+ </condition>
+ <condition property="target.version" value="8" else="11">
+ <equals arg1="${java.version.8}" arg2="true"/>
+ </condition>
+
<!--
Add all the dependencies.
-->
@@@ -550,35 -383,30 +552,35 @@@
<exclusion groupId="org.hamcrest" artifactId="hamcrest-core"/>
</dependency>
<dependency groupId="org.mockito" artifactId="mockito-core" version="3.2.4" scope="test"/>
+ <dependency groupId="org.quicktheories" artifactId="quicktheories" version="0.26" scope="test"/>
+ <dependency groupId="com.google.code.java-allocation-instrumenter" artifactId="java-allocation-instrumenter" version="${allocation-instrumenter.version}" scope="test">
+ <exclusion groupId="com.google.guava" artifactId="guava"/>
+ </dependency>
- <dependency groupId="org.apache.cassandra" artifactId="dtest-api" version="0.0.13" scope="test"/>
+ <dependency groupId="org.apache.cassandra" artifactId="dtest-api" version="0.0.15" scope="test"/>
<dependency groupId="org.reflections" artifactId="reflections" version="0.10.2" scope="test"/>
- <dependency groupId="org.quicktheories" artifactId="quicktheories" version="0.25" scope="test"/>
<dependency groupId="org.apache.hadoop" artifactId="hadoop-core" version="1.0.3" scope="provided">
- <exclusion groupId="org.mortbay.jetty" artifactId="servlet-api"/>
- <exclusion groupId="commons-logging" artifactId="commons-logging"/>
- <exclusion groupId="org.eclipse.jdt" artifactId="core"/>
- <exclusion groupId="ant" artifactId="ant"/>
- <exclusion groupId="junit" artifactId="junit"/>
+ <exclusion groupId="org.mortbay.jetty" artifactId="servlet-api"/>
+ <exclusion groupId="commons-logging" artifactId="commons-logging"/>
+ <exclusion groupId="org.eclipse.jdt" artifactId="core"/>
+ <exclusion groupId="ant" artifactId="ant"/>
+ <exclusion groupId="junit" artifactId="junit"/>
<exclusion groupId="org.codehaus.jackson" artifactId="jackson-mapper-asl"/>
+ <exclusion groupId="org.slf4j" artifactId="slf4j-api"/>
</dependency>
<dependency groupId="org.apache.hadoop" artifactId="hadoop-minicluster" version="1.0.3" scope="provided">
- <exclusion groupId="asm" artifactId="asm"/> <!-- this is the outdated version 3.1 -->
+ <exclusion groupId="asm" artifactId="asm"/> <!-- this is the outdated version 3.1 -->
+ <exclusion groupId="org.slf4j" artifactId="slf4j-api"/>
<exclusion groupId="org.codehaus.jackson" artifactId="jackson-mapper-asl"/>
</dependency>
- <dependency groupId="net.java.dev.jna" artifactId="jna" version="4.2.2"/>
+ <dependency groupId="net.java.dev.jna" artifactId="jna" version="5.6.0"/>
- <dependency groupId="org.jacoco" artifactId="org.jacoco.agent" version="${jacoco.version}"/>
- <dependency groupId="org.jacoco" artifactId="org.jacoco.ant" version="${jacoco.version}"/>
+ <dependency groupId="org.jacoco" artifactId="org.jacoco.agent" version="${jacoco.version}" scope="test"/>
+ <dependency groupId="org.jacoco" artifactId="org.jacoco.ant" version="${jacoco.version}" scope="test"/>
- <dependency groupId="org.jboss.byteman" artifactId="byteman-install" version="${byteman.version}"/>
- <dependency groupId="org.jboss.byteman" artifactId="byteman" version="${byteman.version}"/>
- <dependency groupId="org.jboss.byteman" artifactId="byteman-submit" version="${byteman.version}"/>
- <dependency groupId="org.jboss.byteman" artifactId="byteman-bmunit" version="${byteman.version}"/>
+ <dependency groupId="org.jboss.byteman" artifactId="byteman-install" version="${byteman.version}" scope="provided"/>
+ <dependency groupId="org.jboss.byteman" artifactId="byteman" version="${byteman.version}" scope="provided"/>
+ <dependency groupId="org.jboss.byteman" artifactId="byteman-submit" version="${byteman.version}" scope="provided"/>
+ <dependency groupId="org.jboss.byteman" artifactId="byteman-bmunit" version="${byteman.version}" scope="provided"/>
<dependency groupId="net.bytebuddy" artifactId="byte-buddy" version="${bytebuddy.version}" />
<dependency groupId="net.bytebuddy" artifactId="byte-buddy-agent" version="${bytebuddy.version}" />
@@@ -1300,9 -1117,8 +1302,10 @@@
encoding="utf-8">
<classpath>
<path refid="cassandra.classpath.test"/>
+ <pathelement location="${fqltool.build.classes}"/>
</classpath>
<compilerarg value="-XDignore.symbol.file"/>
++ <compilerarg line="${jdk11-javac-exports}"/>
<src path="${test.anttasks.src}"/>
<src path="${test.unit.src}"/>
<src path="${test.long.src}"/>
@@@ -1998,32 -1753,13 +2001,32 @@@
</java>
</target>
+ <target name="_maybe_update_idea_to_java11" if="java.version.11">
+ <replace file="${eclipse.project.name}.iml" token="JDK_1_8" value="JDK_11"/>
+ <replace file=".idea/misc.xml" token="JDK_1_8" value="JDK_11"/>
+ <replace file=".idea/misc.xml" token="1.8" value="11"/>
+ <replaceregexp file=".idea/workspace.xml"
+ match="name="VM_PARAMETERS" value="(.*)"
+ replace="name="VM_PARAMETERS" value="\1 ${java11-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" />
++ <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>
+ </target>
+
<!-- Generate IDEA project description files -->
- <target name="generate-idea-files" depends="init,maven-ant-tasks-retrieve-build,gen-cql3-grammar,createVersionPropFile" description="Generate IDEA files">
+ <target name="generate-idea-files" depends="init,maven-ant-tasks-init,resolver-dist-lib,gen-cql3-grammar,generate-jflex-java,createVersionPropFile" description="Generate IDEA files">
<mkdir dir=".idea"/>
<mkdir dir=".idea/libraries"/>
- <copy todir=".idea">
+ <copy todir=".idea" overwrite="true">
<fileset dir="ide/idea"/>
</copy>
+ <replace file=".idea/workspace.xml" token="trunk" value="${eclipse.project.name}"/>
+ <replace file=".idea/workspace.xml" token="-Dcassandra.use_nix_recursive_delete=true" value="-Dcassandra.use_nix_recursive_delete=${cassandra.use_nix_recursive_delete}"/>
<copy tofile="${eclipse.project.name}.iml" file="ide/idea-iml-file.xml"/>
<echo file=".idea/.name">Apache Cassandra ${eclipse.project.name}</echo>
<echo file=".idea/modules.xml"><![CDATA[<?xml version="1.0" encoding="UTF-8"?>
diff --cc relocate-dependencies.pom
index 07728dd405,0000000000..96d8d16ee9
mode 100644,000000..100644
--- a/relocate-dependencies.pom
+++ b/relocate-dependencies.pom
@@@ -1,153 -1,0 +1,153 @@@
+<!--
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+-->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <groupId>org.apache.cassandra</groupId>
+ <artifactId>cassandra-dtest-shaded</artifactId>
+ <version>4.0.1-SNAPSHOT</version>
+ <packaging>jar</packaging>
+
+ <name>Cassandra in-jvm dtests shaded jar</name>
+
+ <properties>
+ <project.type>library</project.type>
+ <java.version>1.8</java.version>
+ <test.source.directory>src/test/unit/java</test.source.directory>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+
+ <junit.version>4.12</junit.version>
+ <maven.compiler.source>${java.version}</maven.compiler.source>
+ <maven.compiler.target>${java.version}</maven.compiler.target>
+ <dtest-local.version>4.0.0-SNAPSHOT</dtest-local.version>
+ </properties>
+
+ <dependencies>
+
+ <dependency>
+ <groupId>org.apache.cassandra</groupId>
+ <artifactId>cassandra-dtest-local</artifactId>
+ <version>${dtest-local.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>${junit.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ </dependencies>
+
+ <build>
+ <testSourceDirectory>${test.source.directory}</testSourceDirectory>
+
+ <plugins>
+ <!-- generate a shaded JAR -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
- <version>3.2.1</version>
++ <version>3.4.1</version>
+
+ <configuration>
+ <createSourcesJar>false</createSourcesJar>
+
+ <artifactSet>
+ <includes>
+ <include>org.apache.cassandra:cassandra-dtest-local</include>
+ </includes>
+ </artifactSet>
+
+ <relocations>
+ <relocation>
+ <pattern>io.netty</pattern>
+ <shadedPattern>relocated.shaded.io.netty</shadedPattern>
+ </relocation>
+
+ <relocation>
+ <pattern>com.google</pattern>
+ <shadedPattern>relocated.shaded.com.google</shadedPattern>
+ </relocation>
+
+ <relocation>
+ <pattern>com.datastax</pattern>
+ <shadedPattern>relocated.shaded.com.datastax</shadedPattern>
+ </relocation>
+ </relocations>
+
+ <filters>
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>**/Log4j2Plugins.dat</exclude>
+ </excludes>
+ </filter>
+
+ <filter>
+ <artifact>io.netty:netty-*</artifact>
+ <excludes>
+ <exclude>META-INF/maven/</exclude>
+ <exclude>META-INF/io.netty.versions.properties</exclude>
+ </excludes>
+ </filter>
+
+ <filter>
+ <artifact>com.google.guava:guava</artifact>
+ <excludes>
+ <exclude>META-INF/maven/</exclude>
+ </excludes>
+ </filter>
+
+ <filter>
+ <artifact>com.google.guava:failureaccess</artifact>
+ <excludes>
+ <exclude>META-INF/maven/</exclude>
+ </excludes>
+ </filter>
+
+ <filter>
+ <artifact>com.datastax.cassandra:cassandra-driver-core</artifact>
+ <excludes>
+ <exclude>META-INF/maven/</exclude>
+ </excludes>
+ </filter>
+
+ </filters>
+ </configuration>
+
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+
+ </configuration>
+ </execution>
+ </executions>
+
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --cc src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
index 67de29b999,0000000000..67637ac76e
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
+++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
@@@ -1,385 -1,0 +1,401 @@@
+/*
+ * 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.config;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.cassandra.exceptions.ConfigurationException;
+
+/** A class that extracts system properties for the cassandra node it runs within. */
+public enum CassandraRelevantProperties
+{
+ //base JVM properties
+ JAVA_HOME("java.home"),
+ CASSANDRA_PID_FILE ("cassandra-pidfile"),
+
+ /**
+ * Indicates the temporary directory used by the Java Virtual Machine (JVM)
+ * to create and store temporary files.
+ */
+ JAVA_IO_TMPDIR ("java.io.tmpdir"),
+
+ /**
+ * Path from which to load native libraries.
+ * Default is absolute path to lib directory.
+ */
+ JAVA_LIBRARY_PATH ("java.library.path"),
+
+ JAVA_SECURITY_EGD ("java.security.egd"),
+
+ /** Java Runtime Environment version */
+ JAVA_VERSION ("java.version"),
+
+ /** Java Virtual Machine implementation name */
+ JAVA_VM_NAME ("java.vm.name"),
+
+ /** Line separator ("\n" on UNIX). */
+ LINE_SEPARATOR ("line.separator"),
+
+ /** Java class path. */
+ JAVA_CLASS_PATH ("java.class.path"),
+
+ /** Operating system architecture. */
+ OS_ARCH ("os.arch"),
+
+ /** Operating system name. */
+ OS_NAME ("os.name"),
+
+ /** User's home directory. */
+ USER_HOME ("user.home"),
+
+ /** Platform word size sun.arch.data.model. Examples: "32", "64", "unknown"*/
+ SUN_ARCH_DATA_MODEL ("sun.arch.data.model"),
+
+ //JMX properties
+ /**
+ * The value of this property represents the host name string
+ * that should be associated with remote stubs for locally created remote objects,
+ * in order to allow clients to invoke methods on the remote object.
+ */
+ JAVA_RMI_SERVER_HOSTNAME ("java.rmi.server.hostname"),
+
+ /**
+ * If this value is true, object identifiers for remote objects exported by this VM will be generated by using
+ * a cryptographically secure random number generator. The default value is false.
+ */
+ JAVA_RMI_SERVER_RANDOM_ID ("java.rmi.server.randomIDs"),
+
+ /**
+ * This property indicates whether password authentication for remote monitoring is
+ * enabled. By default it is disabled - com.sun.management.jmxremote.authenticate
+ */
+ 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
+ * if there is no port specified, if the port does not have the correct numeric format,
+ * or if the specified name is empty or null.
+ */
+ COM_SUN_MANAGEMENT_JMXREMOTE_RMI_PORT ("com.sun.management.jmxremote.rmi.port", "0"),
+
+ /** Cassandra jmx remote port */
+ CASSANDRA_JMX_REMOTE_PORT("cassandra.jmx.remote.port"),
+
+ /** This property indicates whether SSL is enabled for monitoring remotely. Default is set to false. */
+ COM_SUN_MANAGEMENT_JMXREMOTE_SSL ("com.sun.management.jmxremote.ssl"),
+
+ /**
+ * This property indicates whether SSL client authentication is enabled - com.sun.management.jmxremote.ssl.need.client.auth.
+ * Default is set to false.
+ */
+ COM_SUN_MANAGEMENT_JMXREMOTE_SSL_NEED_CLIENT_AUTH ("com.sun.management.jmxremote.ssl.need.client.auth"),
+
+ /**
+ * This property indicates the location for the access file. If com.sun.management.jmxremote.authenticate is false,
+ * then this property and the password and access files, are ignored. Otherwise, the access file must exist and
+ * be in the valid format. If the access file is empty or nonexistent, then no access is allowed.
+ */
+ COM_SUN_MANAGEMENT_JMXREMOTE_ACCESS_FILE ("com.sun.management.jmxremote.access.file"),
+
+ /** This property indicates the path to the password file - com.sun.management.jmxremote.password.file */
+ COM_SUN_MANAGEMENT_JMXREMOTE_PASSWORD_FILE ("com.sun.management.jmxremote.password.file"),
+
+ /** Port number to enable JMX RMI connections - com.sun.management.jmxremote.port */
+ COM_SUN_MANAGEMENT_JMXREMOTE_PORT ("com.sun.management.jmxremote.port"),
+
+ /**
+ * A comma-delimited list of SSL/TLS protocol versions to enable.
+ * Used in conjunction with com.sun.management.jmxremote.ssl - com.sun.management.jmxremote.ssl.enabled.protocols
+ */
+ COM_SUN_MANAGEMENT_JMXREMOTE_SSL_ENABLED_PROTOCOLS ("com.sun.management.jmxremote.ssl.enabled.protocols"),
+
+ /**
+ * A comma-delimited list of SSL/TLS cipher suites to enable.
+ * Used in conjunction with com.sun.management.jmxremote.ssl - com.sun.management.jmxremote.ssl.enabled.cipher.suites
+ */
+ COM_SUN_MANAGEMENT_JMXREMOTE_SSL_ENABLED_CIPHER_SUITES ("com.sun.management.jmxremote.ssl.enabled.cipher.suites"),
+
+ /** mx4jaddress */
+ MX4JADDRESS ("mx4jaddress"),
+
+ /** mx4jport */
+ MX4JPORT ("mx4jport"),
+
+ /**
+ * When bootstraping we wait for all schema versions found in gossip to be seen, and if not seen in time we fail
+ * the bootstrap; this property will avoid failing and allow bootstrap to continue if set to true.
+ */
+ BOOTSTRAP_SKIP_SCHEMA_CHECK("cassandra.skip_schema_check"),
+
+ /**
+ * When bootstraping how long to wait for schema versions to be seen.
+ */
+ BOOTSTRAP_SCHEMA_DELAY_MS("cassandra.schema_delay_ms"),
+
+ /**
+ * When draining, how long to wait for mutating executors to shutdown.
+ */
+ DRAIN_EXECUTOR_TIMEOUT_MS("cassandra.drain_executor_timeout_ms", String.valueOf(TimeUnit.MINUTES.toMillis(5))),
+
+ /**
+ * Gossip quarantine delay is used while evaluating membership changes and should only be changed with extreme care.
+ */
+ GOSSIPER_QUARANTINE_DELAY("cassandra.gossip_quarantine_delay_ms"),
+
+ IGNORED_SCHEMA_CHECK_VERSIONS("cassandra.skip_schema_check_for_versions"),
+
+ IGNORED_SCHEMA_CHECK_ENDPOINTS("cassandra.skip_schema_check_for_endpoints"),
+
+ /**
+ * When doing a host replacement its possible that the gossip state is "empty" meaning that the endpoint is known
+ * but the current state isn't known. If the host replacement is needed to repair this state, this property must
+ * be true.
+ */
+ REPLACEMENT_ALLOW_EMPTY("cassandra.allow_empty_replace_address", "true"),
+
+ /**
+ * Whether {@link org.apache.cassandra.db.ConsistencyLevel#NODE_LOCAL} should be allowed.
+ */
+ ENABLE_NODELOCAL_QUERIES("cassandra.enable_nodelocal_queries", "false"),
+
+ //cassandra properties (without the "cassandra." prefix)
+
+ /**
+ * The cassandra-foreground option will tell CassandraDaemon whether
+ * to close stdout/stderr, but it's up to us not to background.
+ * yes/null
+ */
+ CASSANDRA_FOREGROUND ("cassandra-foreground"),
+
+ DEFAULT_PROVIDE_OVERLAPPING_TOMBSTONES ("default.provide.overlapping.tombstones"),
+ ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION ("org.apache.cassandra.disable_mbean_registration"),
+ //only for testing
+ ORG_APACHE_CASSANDRA_CONF_CASSANDRA_RELEVANT_PROPERTIES_TEST("org.apache.cassandra.conf.CassandraRelevantPropertiesTest"),
+ ORG_APACHE_CASSANDRA_DB_VIRTUAL_SYSTEM_PROPERTIES_TABLE_TEST("org.apache.cassandra.db.virtual.SystemPropertiesTableTest"),
+
+ /** This property indicates whether disable_mbean_registration is true */
+ IS_DISABLED_MBEAN_REGISTRATION("org.apache.cassandra.disable_mbean_registration"),
+
+ /** 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"),
++
+ MIGRATION_DELAY("cassandra.migration_delay_ms", "60000"),
+ /** Defines how often schema definitions are pulled from the other nodes */
+ SCHEMA_PULL_INTERVAL_MS("cassandra.schema_pull_interval_ms", "60000"),
+ /**
+ * Minimum delay after a failed pull request before it is reattempted. It prevents reattempting failed requests
+ * immediately as it is high chance they will fail anyway. It is better to wait a bit instead of flooding logs
+ * and wasting resources.
+ */
+ SCHEMA_PULL_BACKOFF_DELAY_MS("cassandra.schema_pull_backoff_delay_ms", "3000"),
+
+ /** When enabled, recursive directory deletion will be executed using a unix command `rm -rf` instead of traversing
+ * and removing individual files. This is now used only tests, but eventually we will make it true by default.*/
+ USE_NIX_RECURSIVE_DELETE("cassandra.use_nix_recursive_delete"),
+
+ /** If set, {@link org.apache.cassandra.net.MessagingService} is shutdown abrtuptly without waiting for anything.
+ * This is an optimization used in unit tests becuase we never restart a node there. The only node is stopoped
+ * when the JVM terminates. Therefore, we can use such optimization and not wait unnecessarily. */
+ NON_GRACEFUL_SHUTDOWN("cassandra.test.messagingService.nonGracefulShutdown"),
+
+ /** Flush changes of {@link org.apache.cassandra.schema.SchemaKeyspace} after each schema modification. In production,
+ * we always do that. However, tests which do not restart nodes may disable this functionality in order to run
+ * faster. Note that this is disabled for unit tests but if an individual test requires schema to be flushed, it
+ * can be also done manually for that particular case: {@code flush(SchemaConstants.SCHEMA_KEYSPACE_NAME);}. */
+ FLUSH_LOCAL_SCHEMA_CHANGES("cassandra.test.flush_local_schema_changes", "true"),
+
+ ;
+
+ CassandraRelevantProperties(String key, String defaultVal)
+ {
+ this.key = key;
+ this.defaultVal = defaultVal;
+ }
+
+ CassandraRelevantProperties(String key)
+ {
+ this.key = key;
+ this.defaultVal = null;
+ }
+
+ private final String key;
+ private final String defaultVal;
+
+ public String getKey()
+ {
+ return key;
+ }
+
+ /**
+ * Gets the value of the indicated system property.
+ * @return system property value if it exists, defaultValue otherwise.
+ */
+ public String getString()
+ {
+ String value = System.getProperty(key);
+
+ return value == null ? defaultVal : STRING_CONVERTER.convert(value);
+ }
+
+ /**
+ * Sets the property to its default value if a default value was specified. Remove the property otherwise.
+ */
+ public void reset()
+ {
+ if (defaultVal != null)
+ System.setProperty(key, defaultVal);
+ else
+ System.getProperties().remove(key);
+ }
+
+ /**
+ * Gets the value of a system property as a String.
+ * @return system property String value if it exists, overrideDefaultValue otherwise.
+ */
+ public String getString(String overrideDefaultValue)
+ {
+ String value = System.getProperty(key);
+ if (value == null)
+ return overrideDefaultValue;
+
+ return STRING_CONVERTER.convert(value);
+ }
+
+ /**
+ * Sets the value into system properties.
+ * @param value to set
+ */
+ public void setString(String value)
+ {
+ System.setProperty(key, value);
+ }
+
+ /**
+ * Gets the value of a system property as a boolean.
+ * @return system property boolean value if it exists, false otherwise().
+ */
+ public boolean getBoolean()
+ {
+ String value = System.getProperty(key);
+
+ return BOOLEAN_CONVERTER.convert(value == null ? defaultVal : value);
+ }
+
+ /**
+ * Gets the value of a system property as a boolean.
+ * @return system property boolean value if it exists, overrideDefaultValue otherwise.
+ */
+ public boolean getBoolean(boolean overrideDefaultValue)
+ {
+ String value = System.getProperty(key);
+ if (value == null)
+ return overrideDefaultValue;
+
+ return BOOLEAN_CONVERTER.convert(value);
+ }
+
+ /**
+ * Sets the value into system properties.
+ * @param value to set
+ */
+ public void setBoolean(boolean value)
+ {
+ System.setProperty(key, Boolean.toString(value));
+ }
+
+ /**
+ * Gets the value of a system property as a int.
+ * @return system property int value if it exists, defaultValue otherwise.
+ */
+ public int getInt()
+ {
+ String value = System.getProperty(key);
+
+ return INTEGER_CONVERTER.convert(value == null ? defaultVal : value);
+ }
+
+ /**
+ * Gets the value of a system property as a int.
+ * @return system property int value if it exists, overrideDefaultValue otherwise.
+ */
+ public int getInt(int overrideDefaultValue)
+ {
+ String value = System.getProperty(key);
+ if (value == null)
+ return overrideDefaultValue;
+
+ return INTEGER_CONVERTER.convert(value);
+ }
+
+ /**
+ * Sets the value into system properties.
+ * @param value to set
+ */
+ public void setInt(int value)
+ {
+ System.setProperty(key, Integer.toString(value));
+ }
+
+ private interface PropertyConverter<T>
+ {
+ T convert(String value);
+ }
+
+ private static final PropertyConverter<String> STRING_CONVERTER = value -> value;
+
+ private static final PropertyConverter<Boolean> BOOLEAN_CONVERTER = Boolean::parseBoolean;
+
+ private static final PropertyConverter<Integer> INTEGER_CONVERTER = value ->
+ {
+ try
+ {
+ return Integer.decode(value);
+ }
+ catch (NumberFormatException e)
+ {
+ throw new ConfigurationException(String.format("Invalid value for system property: " +
+ "expected integer value but got '%s'", value));
+ }
+ };
+
+ /**
+ * @return whether a system property is present or not.
+ */
+ public boolean isPresent()
+ {
+ return System.getProperties().containsKey(key);
+ }
+}
diff --cc src/java/org/apache/cassandra/service/GCInspector.java
index 0d42e72425,0c0f9e4704..3154d89427
--- a/src/java/org/apache/cassandra/service/GCInspector.java
+++ b/src/java/org/apache/cassandra/service/GCInspector.java
@@@ -157,10 -146,10 +155,10 @@@ public class GCInspector implements Not
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, me);
+ MBeanWrapper.instance.registerMBean(this, new ObjectName(MBEAN_NAME));
}
- catch (Exception e)
+ catch (MalformedObjectNameException | IOException e)
{
throw new RuntimeException(e);
}
diff --cc src/java/org/apache/cassandra/utils/Isolated.java
index 898631ff42,0000000000..12f88e3395
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/utils/Isolated.java
+++ b/src/java/org/apache/cassandra/utils/Isolated.java
@@@ -1,39 -1,0 +1,41 @@@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
- package org.apache.cassandra.distributed.shared;
++package org.apache.cassandra.utils;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
++import org.apache.cassandra.utils.Shared;
++
+/**
+ * Tells jvm-dtest that a class should be isolated and loaded into the instance class loader.
+ *
+ * Jvm-dtest relies on classloader isolation to run multiple cassandra instances in the same JVM, this makes it
+ * so some classes do not get shared (outside a blesssed set of classes/packages). When the default behavior
+ * is not desirable, this annotation will tell jvm-dtest to isolate the class accross all class loaders.
+ *
+ * This is the oposite of {@link Shared}.
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ ElementType.TYPE })
+public @interface Isolated
+{
+}
diff --cc src/java/org/apache/cassandra/utils/JMXServerUtils.java
index 5557fda345,aad4f0512c..49ff5a07d2
--- a/src/java/org/apache/cassandra/utils/JMXServerUtils.java
+++ b/src/java/org/apache/cassandra/utils/JMXServerUtils.java
@@@ -325,7 -287,8 +328,7 @@@ public class JMXServerUtil
* 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
- {
++ public static class JmxRegistry extends sun.rmi.registry.RegistryImpl {
private final String lookupName;
private Remote remoteServerStub;
diff --cc src/java/org/apache/cassandra/utils/MBeanWrapper.java
index f8bc439d7b,a76a2e7029..0ef342da30
--- a/src/java/org/apache/cassandra/utils/MBeanWrapper.java
+++ b/src/java/org/apache/cassandra/utils/MBeanWrapper.java
@@@ -27,32 -32,69 +32,56 @@@ 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;
+
/**
* Helper class to avoid catching and rethrowing checked exceptions on MBean and
* allow turning of MBean registration for test purposes.
*/
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();
- String IS_DISABLED_MBEAN_REGISTRATION = "org.apache.cassandra.disable_mbean_registration";
- String DTEST_IS_IN_JVM_DTEST = "org.apache.cassandra.dtest.is_in_jvm_dtest";
- String MBEAN_REGISTRATION_CLASS = "org.apache.cassandra.mbean_registration_class";
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 (Boolean.getBoolean(DTEST_IS_IN_JVM_DTEST))
++ if (DTEST_IS_IN_JVM_DTEST.getBoolean())
+ {
+ return new DelegatingMbeanWrapper(getMBeanWrapper());
+ }
+
+ return getMBeanWrapper();
+ }
+
+ static MBeanWrapper getMBeanWrapper()
+ {
- if (Boolean.getBoolean(IS_DISABLED_MBEAN_REGISTRATION))
++ if (ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION.getBoolean())
+ {
return new NoOpMBeanWrapper();
+ }
- String klass = System.getProperty(MBEAN_REGISTRATION_CLASS);
+ String klass = MBEAN_REGISTRATION_CLASS.getString();
if (klass == null)
- return new PlatformMBeanWrapper();
+ {
- if (Boolean.getBoolean(DTEST_IS_IN_JVM_DTEST))
++ if (DTEST_IS_IN_JVM_DTEST.getBoolean())
+ {
+ return new NoOpMBeanWrapper();
+ }
+ else
+ {
+ return new PlatformMBeanWrapper();
+ }
+ }
return FBUtilities.construct(klass, "mbean");
}
- static ObjectName create(String mbeanName, OnException onException)
- {
- try
- {
- return new ObjectName(mbeanName);
- }
- catch (MalformedObjectNameException e)
- {
- onException.handler.accept(e);
- return null;
- }
- }
-
// 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,15 -104,19 +91,17 @@@
{
ObjectName name = create(mbeanName, onException);
if (name == null)
+ {
return;
+ }
registerMBean(obj, name, onException);
}
-
default void registerMBean(Object obj, String mbeanName)
{
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,15 -126,19 +111,17 @@@
{
ObjectName name = create(mbeanName, onException);
if (name == null)
+ {
return false;
+ }
return isRegistered(name, onException);
}
-
default boolean isRegistered(String mbeanName)
{
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,9 -148,12 +131,11 @@@
{
ObjectName name = create(mbeanName, onException);
if (name == null)
+ {
return;
+ }
unregisterMBean(name, onException);
}
-
default void unregisterMBean(String mbeanName)
{
unregisterMBean(mbeanName, OnException.THROW);
@@@ -123,15 -180,18 +162,17 @@@
{
public void registerMBean(Object obj, ObjectName mbeanName, OnException onException) {}
public void registerMBean(Object obj, String mbeanName, OnException onException) {}
- public boolean isRegistered(ObjectName mbeanName, OnException onException) { return false;}
- public boolean isRegistered(String mbeanName, OnException onException) { return false;}
+ public boolean isRegistered(ObjectName mbeanName, OnException onException) { return false; }
+ 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)
{
try
@@@ -168,18 -228,160 +209,167 @@@
onException.handler.accept(e);
}
}
+
+ public Set<ObjectName> queryNames(ObjectName name, QueryExp query)
+ {
+ return mbs.queryNames(name, query);
+ }
+
+ public MBeanServer getMBeanServer()
+ {
+ return mbs;
+ }
-
+ }
+
+ 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()
- {
++ public void close() {
+ mbs.queryNames(null, null).forEach(name -> {
- try
- {
++ try {
+ if (!name.getCanonicalName().contains("MBeanServerDelegate"))
+ {
+ mbs.unregisterMBean(name);
+ }
- }
- catch (Throwable e)
- {
++ } 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 MBeanWrapper getDelegate()
- {
- return delegate;
++ public void setDelegate(MBeanWrapper wrapper) {
++ delegate = wrapper;
+ }
+
- public void setDelegate(MBeanWrapper wrapper)
++ public MBeanWrapper getDelegate()
+ {
- delegate = wrapper;
++ 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();
+ }
}
+
- public enum OnException
++ enum OnException
+ {
+ THROW(e -> { throw new RuntimeException(e); }),
+ LOG(e -> { logger.error("Error in MBean wrapper: ", e); }),
+ IGNORE(e -> {});
+
+ public final Consumer<Exception> handler;
+ OnException(Consumer<Exception> handler)
+ {
+ this.handler = handler;
+ }
+ }
}
diff --cc src/java/org/apache/cassandra/utils/NativeLibrary.java
index e5b5da7308,c333794ad8..c17a87cc43
--- a/src/java/org/apache/cassandra/utils/NativeLibrary.java
+++ b/src/java/org/apache/cassandra/utils/NativeLibrary.java
@@@ -39,10 -38,10 +39,12 @@@ import static org.apache.cassandra.util
import static org.apache.cassandra.utils.NativeLibrary.OSType.WINDOWS;
import static org.apache.cassandra.utils.NativeLibrary.OSType.AIX;
++@Shared
public final class NativeLibrary
{
private static final Logger logger = LoggerFactory.getLogger(NativeLibrary.class);
++ @Shared
public enum OSType
{
LINUX,
diff --cc src/java/org/apache/cassandra/utils/NativeLibraryDarwin.java
index 6ed18d13a4,6ed18d13a4..c119311370
--- a/src/java/org/apache/cassandra/utils/NativeLibraryDarwin.java
+++ b/src/java/org/apache/cassandra/utils/NativeLibraryDarwin.java
@@@ -42,6 -42,6 +42,7 @@@ import com.sun.jna.Pointer
* @see org.apache.cassandra.utils.NativeLibraryWrapper
* @see NativeLibrary
*/
++@Shared
public class NativeLibraryDarwin implements NativeLibraryWrapper
{
private static final Logger logger = LoggerFactory.getLogger(NativeLibraryDarwin.class);
diff --cc src/java/org/apache/cassandra/utils/NativeLibraryLinux.java
index 3f21d17ca3,3f21d17ca3..9c7bb3b73b
--- a/src/java/org/apache/cassandra/utils/NativeLibraryLinux.java
+++ b/src/java/org/apache/cassandra/utils/NativeLibraryLinux.java
@@@ -42,6 -42,6 +42,7 @@@ import com.sun.jna.Pointer
* @see org.apache.cassandra.utils.NativeLibraryWrapper
* @see NativeLibrary
*/
++@Shared
public class NativeLibraryLinux implements NativeLibraryWrapper
{
private static boolean available;
diff --cc src/java/org/apache/cassandra/utils/NativeLibraryWindows.java
index b8304c791f,b8304c791f..85872ab50e
--- a/src/java/org/apache/cassandra/utils/NativeLibraryWindows.java
+++ b/src/java/org/apache/cassandra/utils/NativeLibraryWindows.java
@@@ -35,6 -35,6 +35,7 @@@ import com.sun.jna.Pointer
* @see org.apache.cassandra.utils.NativeLibraryWrapper
* @see NativeLibrary
*/
++@Shared
public class NativeLibraryWindows implements NativeLibraryWrapper
{
private static final Logger logger = LoggerFactory.getLogger(NativeLibraryWindows.class);
diff --cc src/java/org/apache/cassandra/utils/NativeLibraryWrapper.java
index 879ea88605,879ea88605..45bf4c94d0
--- a/src/java/org/apache/cassandra/utils/NativeLibraryWrapper.java
+++ b/src/java/org/apache/cassandra/utils/NativeLibraryWrapper.java
@@@ -24,6 -24,6 +24,7 @@@ import com.sun.jna.Pointer
* An interface to implement for using OS specific native methods.
* @see NativeLibrary
*/
++@Shared
interface NativeLibraryWrapper
{
/**
diff --cc src/java/org/apache/cassandra/utils/ReflectionUtils.java
index 0000000000,0000000000..bd605db3bc
new file mode 100644
--- /dev/null
+++ b/src/java/org/apache/cassandra/utils/ReflectionUtils.java
@@@ -1,0 -1,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 --cc src/java/org/apache/cassandra/utils/Shared.java
index bb67070c20,0000000000..880acc2fd6
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/utils/Shared.java
+++ b/src/java/org/apache/cassandra/utils/Shared.java
@@@ -1,39 -1,0 +1,39 @@@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
- package org.apache.cassandra.distributed.shared;
++package org.apache.cassandra.utils;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Tells jvm-dtest that a class should be shared accross all {@link ClassLoader}s.
+ *
+ * Jvm-dtest relies on classloader isolation to run multiple cassandra instances in the same JVM, this makes it
+ * so some classes do not get shared (outside a blesssed set of classes/packages). When the default behavior
+ * is not desirable, this annotation will tell jvm-dtest to share the class accross all class loaders.
+ *
+ * This is the oposite of {@link Isolated}.
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ ElementType.TYPE })
+public @interface Shared
+{
+}
diff --cc test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
index 95a2ac5a62,cffbbc461a..fcbef4c220
--- a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
@@@ -69,22 -63,21 +69,22 @@@ import org.apache.cassandra.distributed
import org.apache.cassandra.distributed.api.LogAction;
import org.apache.cassandra.distributed.api.NodeToolResult;
import org.apache.cassandra.distributed.api.TokenSupplier;
-import org.apache.cassandra.distributed.shared.AbstractBuilder;
import org.apache.cassandra.distributed.shared.InstanceClassLoader;
- import org.apache.cassandra.distributed.shared.Isolated;
import org.apache.cassandra.distributed.shared.MessageFilters;
import org.apache.cassandra.distributed.shared.Metrics;
import org.apache.cassandra.distributed.shared.NetworkTopology;
--import org.apache.cassandra.distributed.shared.Shared;
import org.apache.cassandra.distributed.shared.ShutdownException;
import org.apache.cassandra.distributed.shared.Versions;
import org.apache.cassandra.io.util.FileUtils;
-import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.net.Verb;
import org.apache.cassandra.utils.FBUtilities;
++import org.apache.cassandra.utils.Isolated;
++import org.apache.cassandra.utils.Shared;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
import org.reflections.Reflections;
+import org.reflections.util.ConfigurationBuilder;
-import static org.apache.cassandra.utils.MBeanWrapper.DTEST_IS_IN_JVM_DTEST;
+import static org.apache.cassandra.distributed.shared.NetworkTopology.addressAndPort;
/**
* AbstractCluster creates, initializes and manages Cassandra instances ({@link Instance}.
@@@ -156,32 -142,11 +156,34 @@@ public abstract class AbstractCluster<
private volatile BiPredicate<Integer, Throwable> ignoreUncaughtThrowable = null;
private final List<Throwable> uncaughtExceptions = new CopyOnWriteArrayList<>();
- private volatile Thread.UncaughtExceptionHandler previousHandler = null;
-
+ /**
+ * Common builder, add methods that are applicable to both Cluster and Upgradable cluster here.
+ */
+ public static abstract class AbstractBuilder<I extends IInstance, C extends ICluster, B extends AbstractBuilder<I, C, B>>
+ extends org.apache.cassandra.distributed.shared.AbstractBuilder<I, C, B>
{
- System.setProperty(DTEST_IS_IN_JVM_DTEST, "true");
+ private INodeProvisionStrategy.Strategy nodeProvisionStrategy = INodeProvisionStrategy.Strategy.MultipleNetworkInterfaces;
+
+ {
++ // 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.FLUSH_LOCAL_SCHEMA_CHANGES.reset();
+ CassandraRelevantProperties.NON_GRACEFUL_SHUTDOWN.reset();
+ }
+
+ public AbstractBuilder(Factory<I, C, B> factory)
+ {
+ super(factory);
+ }
+
+ public B withNodeProvisionStrategy(INodeProvisionStrategy.Strategy nodeProvisionStrategy)
+ {
+ this.nodeProvisionStrategy = nodeProvisionStrategy;
+ return (B) this;
+ }
}
+
protected class Wrapper extends DelegatingInvokableInstance implements IUpgradeableInstance
{
private final int generation;
@@@ -503,6 -418,6 +505,16 @@@
return instanceMap.get(addr);
}
++ public List<I> get(int... nodes)
++ {
++ if (nodes == null || nodes.length == 0)
++ throw new IllegalArgumentException("No nodes provided");
++ List<I> 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 --cc test/distributed/org/apache/cassandra/distributed/impl/CollectingRMIServerSocketFactoryImpl.java
index 0000000000,5e67eafef6..0f16286b4d
mode 000000,100644..100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/CollectingRMIServerSocketFactoryImpl.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/CollectingRMIServerSocketFactoryImpl.java
@@@ -1,0 -1,87 +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;
+
++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);
+ }
+ }
diff --cc test/distributed/org/apache/cassandra/distributed/impl/INodeProvisionStrategy.java
index 32f82c0f2f,0000000000..6e8f9ffe26
mode 100644,000000..100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/INodeProvisionStrategy.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/INodeProvisionStrategy.java
@@@ -1,99 -1,0 +1,110 @@@
+/*
+ * 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;
+
+public interface INodeProvisionStrategy
+{
+
+ public enum Strategy
+ {
+ OneNetworkInterface
+ {
+ INodeProvisionStrategy create(int subnet) {
+ return new INodeProvisionStrategy()
+ {
+ public String seedIp()
+ {
+ return "127.0." + subnet + ".1";
+ }
+
+ public int seedPort()
+ {
+ return 7012;
+ }
+
+ public String ipAddress(int nodeNum)
+ {
+ return "127.0." + subnet + ".1";
+ }
+
+ public int storagePort(int nodeNum)
+ {
+ return 7011 + nodeNum;
+ }
+
+ public int nativeTransportPort(int nodeNum)
+ {
+ return 9041 + nodeNum;
+ }
++
++ public int jmxPort(int nodeNum)
++ {
++ return 7199 + nodeNum;
++ }
+ };
+ }
+ },
+ MultipleNetworkInterfaces
+ {
+ INodeProvisionStrategy create(int subnet) {
+ String ipPrefix = "127.0." + subnet + ".";
+ return new INodeProvisionStrategy()
+ {
+ public String seedIp()
+ {
+ return ipPrefix + "1";
+ }
+
+ public int seedPort()
+ {
+ return 7012;
+ }
+
+ public String ipAddress(int nodeNum)
+ {
+ return ipPrefix + nodeNum;
+ }
+
+ public int storagePort(int nodeNum)
+ {
+ return 7012;
+ }
+
+ public int nativeTransportPort(int nodeNum)
+ {
+ return 9042;
+ }
++
++ public int jmxPort(int nodeNum)
++ {
++ return 7199;
++ }
+ };
+ }
+ };
+ abstract INodeProvisionStrategy create(int subnet);
+ }
+
+ abstract String seedIp();
+ abstract int seedPort();
+ abstract String ipAddress(int nodeNum);
+ abstract int storagePort(int nodeNum);
+ abstract int nativeTransportPort(int nodeNum);
++ abstract int jmxPort(int nodeNum);
+}
diff --cc test/distributed/org/apache/cassandra/distributed/impl/Instance.java
index b8833dd319,cca6296629..12ef7a53d4
--- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
@@@ -42,9 -43,7 +42,11 @@@ import java.util.concurrent.atomic.Atom
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;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@@ -130,28 -118,24 +132,40 @@@ import org.apache.cassandra.utils.ByteA
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.UUIDGen;
+import org.apache.cassandra.utils.UUIDSerializer;
import org.apache.cassandra.utils.concurrent.Ref;
-import org.apache.cassandra.utils.memory.BufferPool;
+import org.apache.cassandra.utils.memory.BufferPools;
+import org.apache.cassandra.utils.progress.jmx.JMXBroadcastExecutor;
import static java.util.concurrent.TimeUnit.MINUTES;
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;
+import static org.apache.cassandra.distributed.impl.DistributedTestSnitch.toCassandraInetAddressAndPort;
+import static org.apache.cassandra.net.Verb.BATCH_STORE_REQ;
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;
// should never be invoked directly, so that it is instantiated on other class loader;
// only visible for inheritance
@@@ -498,10 -522,12 +512,13 @@@
mkdirs();
- assert config.networkTopology().contains(config.broadcastAddress());
+ assert config.networkTopology().contains(config.broadcastAddress()) : String.format("Network topology %s doesn't contain the address %s",
+ config.networkTopology(), config.broadcastAddress());
DistributedTestSnitch.assign(config.networkTopology());
+ if (config.has(JMX))
+ startJmx();
+
DatabaseDescriptor.daemonInitialization();
FileUtils.setFSErrorHandler(new DefaultFSErrorHandler());
DatabaseDescriptor.createAllDirectories();
@@@ -624,33 -614,20 +641,47 @@@
initialized = true;
}
+ private void startJmx()
+ {
+ isolatedJmx = new IsolatedJmx(this, inInstancelogger);
+ isolatedJmx.startJmx();
+ }
+
- private void stopJmx() throws IllegalAccessException, NoSuchFieldException, InterruptedException
++ private void stopJmx() throws NoSuchFieldException, InterruptedException, IllegalAccessException
+ {
+ if (config.has(JMX))
+ {
+ isolatedJmx.stopJmx();
+ }
+ }
+
+ // Update the messaging versions for all instances
+ // that have initialized their configurations.
+ private static void propagateMessagingVersions(ICluster cluster)
+ {
+ cluster.stream().forEach(reportToObj -> {
+ IInstance reportTo = (IInstance) reportToObj;
+ if (reportTo.isShutdown())
+ return;
+
+ int reportToVersion = reportTo.getMessagingVersion();
+ if (reportToVersion == 0)
+ return;
+
+ cluster.stream().forEach(reportFromObj -> {
+ IInstance reportFrom = (IInstance) reportFromObj;
+ if (reportFrom == reportTo || reportFrom.isShutdown())
+ return;
+
+ int reportFromVersion = reportFrom.getMessagingVersion();
+ if (reportFromVersion == 0) // has not read configuration yet, no accessing messaging version
+ return;
+ // TODO: decide if we need to take care of the minversion
+ reportTo.setMessagingVersion(reportFrom.broadcastAddress(), reportFromVersion);
+ });
+ });
+ }
+
private void mkdirs()
{
new File(config.getString("saved_caches_directory")).mkdirs();
@@@ -787,34 -763,26 +818,36 @@@
() -> SecondaryIndexManager.shutdownAndWait(1L, MINUTES),
() -> IndexSummaryManager.instance.shutdownAndWait(1L, MINUTES),
() -> ColumnFamilyStore.shutdownExecutorsAndWait(1L, MINUTES),
- () -> PendingRangeCalculatorService.instance.shutdownExecutor(1L, MINUTES),
- () -> BufferPool.shutdownLocalCleaner(1L, MINUTES),
+ () -> PendingRangeCalculatorService.instance.shutdownAndWait(1L, MINUTES),
+ () -> BufferPools.shutdownLocalCleaner(1L, MINUTES),
() -> Ref.shutdownReferenceReaper(1L, MINUTES),
() -> Memtable.MEMORY_POOL.shutdownAndWait(1L, MINUTES),
+ () -> DiagnosticSnapshotService.instance.shutdownAndWait(1L, MINUTES),
() -> SSTableReader.shutdownBlocking(1L, MINUTES),
- () -> DiagnosticSnapshotService.instance.shutdownAndWait(1L, MINUTES)
+ () -> shutdownAndWait(Collections.singletonList(ActiveRepairService.repairCommandExecutor()))
);
+
+ internodeMessagingStarted = false;
error = parallelRun(error, executor,
- () -> ScheduledExecutors.shutdownAndWait(1L, MINUTES),
- (IgnoreThrowingRunnable) MessagingService.instance()::shutdown
+ // can only shutdown message once, so if the test shutsdown an instance, then ignore the failure
+ (IgnoreThrowingRunnable) () -> MessagingService.instance().shutdown(1L, MINUTES, false, true)
);
+
error = parallelRun(error, executor,
- () -> StageManager.shutdownAndWait(1L, MINUTES),
+ () -> GlobalEventExecutor.INSTANCE.awaitInactivity(1L, MINUTES),
+ () -> Stage.shutdownAndWait(1L, MINUTES),
() -> SharedExecutorPool.SHARED.shutdownAndWait(1L, MINUTES)
);
- error = parallelRun(error, executor,
- CommitLog.instance::shutdownBlocking
- );
+ // CommitLog must shut down after Stage, or threads from the latter may attempt to use the former.
+ // (ex. A Mutation stage thread may attempt to add a mutation to the CommitLog.)
+ error = parallelRun(error, executor, CommitLog.instance::shutdownBlocking);
+ error = parallelRun(error, executor, () -> shutdownAndWait(Collections.singletonList(JMXBroadcastExecutor.executor)));
+
+ // ScheduledExecutors shuts down after MessagingService, as MessagingService may issue tasks to it.
+ error = parallelRun(error, executor, () -> ScheduledExecutors.shutdownAndWait(1L, MINUTES));
++
+ error = parallelRun(error, executor, this::stopJmx);
Throwables.maybeFail(error);
}).apply(isolatedExecutor);
diff --cc test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
index 037c221a37,7f6af4b260..06229a0c01
--- a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
@@@ -35,10 -35,9 +35,10 @@@ import com.vdurmont.semver4j.Semver
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IInstanceConfig;
import org.apache.cassandra.distributed.shared.NetworkTopology;
--import org.apache.cassandra.distributed.shared.Shared;
+import org.apache.cassandra.distributed.upgrade.UpgradeTestBase;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.SimpleSeedProvider;
++import org.apache.cassandra.utils.Shared;
@Shared
public class InstanceConfig implements IInstanceConfig
@@@ -76,8 -75,7 +78,9 @@@
String hints_directory,
String cdc_raw_directory,
String initial_token,
+ int storage_port,
- int native_transport_port)
++ int native_transport_port,
+ int jmx_port)
{
this.num = num;
this.networkTopology = networkTopology;
@@@ -270,8 -263,7 +281,9 @@@
String.format("%s/node%d/hints", root, nodeNum),
String.format("%s/node%d/cdc", root, nodeNum),
token,
- 7199);
+ provisionStrategy.storagePort(nodeNum),
- provisionStrategy.nativeTransportPort(nodeNum));
++ provisionStrategy.nativeTransportPort(nodeNum),
++ provisionStrategy.jmxPort(nodeNum));
}
private static String[] datadirs(int datadirCount, File root, int nodeNum)
diff --cc test/distributed/org/apache/cassandra/distributed/impl/IsolatedJmx.java
index 0000000000,b3d06590a9..e19e29fdb3
mode 000000,100644..100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/IsolatedJmx.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/IsolatedJmx.java
@@@ -1,0 -1,231 +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;
-import static org.apache.cassandra.utils.MBeanWrapper.IS_DISABLED_MBEAN_REGISTRATION;
+
+ public class IsolatedJmx
+ {
- /** Controls the JMX server threadpool keap-alive time. */
- private static final String SUN_RMI_TRANSPORT_TCP_THREADKEEPALIVETIME = "sun.rmi.transport.tcp.threadKeepAliveTime";
- /** Controls the distributed garbage collector lease time for JMX objects. */
- private static final String JAVA_RMI_DGC_LEASE_VALUE_IN_JVM_DTEST ="java.rmi.dgc.leaseValue";
+ private static final int RMI_KEEPALIVE_TIME = 1000;
+
+ 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(Instance instance, Logger inInstancelogger) {
- this.inInstancelogger = inInstancelogger;
- config = instance.config();
++ public IsolatedJmx(IInstance instance, Logger inInstanceLogger) {
++ this.config = instance.config();
++ this.inInstancelogger = inInstanceLogger;
+ }
+
- public void startJmx()
- {
++ 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.
- System.setProperty(JAVA_RMI_DGC_LEASE_VALUE_IN_JVM_DTEST, String.valueOf(RMI_KEEPALIVE_TIME));
- System.setProperty(SUN_RMI_TRANSPORT_TCP_THREADKEEPALIVETIME, String.valueOf(RMI_KEEPALIVE_TIME));
- System.setProperty(IS_DISABLED_MBEAN_REGISTRATION, "false");
++ 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 e)
++ catch (Throwable t)
+ {
- throw new RuntimeException("Feature.JMX was enabled but could not be started.", e);
++ throw new RuntimeException("Feature.JMX was enabled but could not be started.", t);
+ }
+ }
+
-
+ private void waitForJmxAvailability(String hostname, int rmiPort, Map<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 = clazz.getDeclaredField(mapName);
++ 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();
+ }
+ }
+ }
+ }
diff --cc test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
index 1755857812,0000000000..2dfcf3f080
mode 100644,000000..100644
--- a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
+++ b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
@@@ -1,805 -1,0 +1,806 @@@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.shared;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import com.google.common.util.concurrent.Futures;
+import org.junit.Assert;
+
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.impl.AbstractCluster;
+import org.apache.cassandra.distributed.impl.InstanceConfig;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.utils.FBUtilities;
++import org.apache.cassandra.utils.Isolated;
+
+import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
+import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Utilities for working with jvm-dtest clusters.
+ *
+ * This class is marked as Isolated as it relies on lambdas, which are in a package that is marked as shared, so need to
+ * tell jvm-dtest to not share this class.
+ *
+ * This class should never be called from within the cluster, always in the App ClassLoader.
+ */
+@Isolated
+public class ClusterUtils
+{
+ /**
+ * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+ */
+ public static <I extends IInstance> I start(I inst, Consumer<WithProperties> fn)
+ {
+ return start(inst, (ignore, prop) -> fn.accept(prop));
+ }
+
+ /**
+ * Start the instance with the given System Properties, after the instance has started, the properties will be cleared.
+ */
+ public static <I extends IInstance> I start(I inst, BiConsumer<I, WithProperties> fn)
+ {
+ try (WithProperties properties = new WithProperties())
+ {
+ fn.accept(inst, properties);
+ inst.startup();
+ return inst;
+ }
+ }
+
+ /**
+ * Stop an instance in a blocking manner.
+ *
+ * The main difference between this and {@link IInstance#shutdown()} is that the wait on the future will catch
+ * the exceptions and throw as runtime.
+ */
+ public static void stopUnchecked(IInstance i)
+ {
+ Futures.getUnchecked(i.shutdown());
+ }
+
+ /**
+ * Stops an instance abruptly. This is done by blocking all messages to/from so all other instances are unable
+ * to communicate, then stopping the instance gracefully.
+ *
+ * The assumption is that hard stopping inbound and outbound messages will apear to the cluster as if the instance
+ * was stopped via kill -9; this does not hold true if the instance is restarted as it knows it was properly shutdown.
+ *
+ * @param cluster to filter messages to
+ * @param inst to shut down
+ */
+ public static <I extends IInstance> void stopAbrupt(ICluster<I> cluster, I inst)
+ {
+ // block all messages to/from the node going down to make sure a clean shutdown doesn't happen
+ IMessageFilters.Filter to = cluster.filters().allVerbs().to(inst.config().num()).drop();
+ IMessageFilters.Filter from = cluster.filters().allVerbs().from(inst.config().num()).drop();
+ try
+ {
+ stopUnchecked(inst);
+ }
+ finally
+ {
+ from.off();
+ to.off();
+ }
+ }
+
+ /**
+ * Stop all the instances in the cluster. This function is differe than {@link ICluster#close()} as it doesn't
+ * clean up the cluster state, it only stops all the instances.
+ */
+ public static <I extends IInstance> void stopAll(ICluster<I> cluster)
+ {
+ cluster.stream().forEach(ClusterUtils::stopUnchecked);
+ }
+
+ /**
+ * Create a new instance and add it to the cluster, without starting it.
+ *
+ * @param cluster to add to
+ * @param dc the instance should be in
+ * @param rack the instance should be in
+ * @param <I> instance type
+ * @return the instance added
+ */
+ public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+ String dc, String rack)
+ {
+ return addInstance(cluster, dc, rack, ignore -> {});
+ }
+
+ /**
+ * Create a new instance and add it to the cluster, without starting it.
+ *
+ * @param cluster to add to
+ * @param dc the instance should be in
+ * @param rack the instance should be in
+ * @param fn function to add to the config before starting
+ * @param <I> instance type
+ * @return the instance added
+ */
+ public static <I extends IInstance> I addInstance(AbstractCluster<I> cluster,
+ String dc, String rack,
+ Consumer<IInstanceConfig> fn)
+ {
+ Objects.requireNonNull(dc, "dc");
+ Objects.requireNonNull(rack, "rack");
+
+ InstanceConfig config = cluster.newInstanceConfig();
+ //TODO adding new instances should be cleaner, currently requires you create the cluster with all
+ // instances known about (at least to NetworkTopology and TokenStategy)
+ // this is very hidden, so should be more explicit
+ config.networkTopology().put(config.broadcastAddress(), NetworkTopology.dcAndRack(dc, rack));
+
+ fn.accept(config);
+
+ return cluster.bootstrap(config);
+ }
+
+ /**
+ * Create and start a new instance that replaces an existing instance.
+ *
+ * The instance will be in the same datacenter and rack as the existing instance.
+ *
+ * @param cluster to add to
+ * @param toReplace instance to replace
+ * @param <I> instance type
+ * @return the instance added
+ */
+ public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster, IInstance toReplace)
+ {
+ return replaceHostAndStart(cluster, toReplace, ignore -> {});
+ }
+
+ /**
+ * Create and start a new instance that replaces an existing instance.
+ *
+ * The instance will be in the same datacenter and rack as the existing instance.
+ *
+ * @param cluster to add to
+ * @param toReplace instance to replace
+ * @param fn lambda to add additional properties
+ * @param <I> instance type
+ * @return the instance added
+ */
+ public static <I extends IInstance> I replaceHostAndStart(AbstractCluster<I> cluster,
+ IInstance toReplace,
+ Consumer<WithProperties> fn)
+ {
+ IInstanceConfig toReplaceConf = toReplace.config();
+ I inst = addInstance(cluster, toReplaceConf.localDatacenter(), toReplaceConf.localRack(), c -> c.set("auto_bootstrap", true));
+
+ return start(inst, properties -> {
+ // lower this so the replacement waits less time
+ properties.setProperty("cassandra.broadcast_interval_ms", Long.toString(TimeUnit.SECONDS.toMillis(30)));
+ // default is 30s, lowering as it should be faster
+ properties.setProperty("cassandra.ring_delay_ms", Long.toString(TimeUnit.SECONDS.toMillis(10)));
+ properties.set(BOOTSTRAP_SCHEMA_DELAY_MS, TimeUnit.SECONDS.toMillis(10));
+
+ // state which node to replace
+ properties.setProperty("cassandra.replace_address_first_boot", toReplace.config().broadcastAddress().getAddress().getHostAddress());
+
+ fn.accept(properties);
+ });
+ }
+
+ /**
+ * Calls {@link org.apache.cassandra.locator.TokenMetadata#sortedTokens()}, returning as a list of strings.
+ */
+ public static List<String> getTokenMetadataTokens(IInvokableInstance inst)
+ {
+ return inst.callOnInstance(() ->
+ StorageService.instance.getTokenMetadata()
+ .sortedTokens().stream()
+ .map(Object::toString)
+ .collect(Collectors.toList()));
+ }
+
+ public static String getLocalToken(IInvokableInstance inst)
+ {
+ return inst.callOnInstance(() -> {
+ List<String> tokens = new ArrayList<>();
+ for (Token t : StorageService.instance.getTokenMetadata().getTokens(FBUtilities.getBroadcastAddressAndPort()))
+ tokens.add(t.getTokenValue().toString());
+
+ assert tokens.size() == 1 : "getLocalToken assumes a single token, but multiple tokens found";
+ return tokens.get(0);
+ });
+ }
+
+ public static <I extends IInstance> void runAndWaitForLogs(Runnable r, String waitString, AbstractCluster<I> cluster) throws TimeoutException
+ {
+ runAndWaitForLogs(r, waitString, cluster.stream().toArray(IInstance[]::new));
+ }
+
+ public static void runAndWaitForLogs(Runnable r, String waitString, IInstance...instances) throws TimeoutException
+ {
+ long [] marks = new long[instances.length];
+ for (int i = 0; i < instances.length; i++)
+ marks[i] = instances[i].logs().mark();
+ r.run();
+ for (int i = 0; i < instances.length; i++)
+ instances[i].logs().watchFor(marks[i], waitString);
+ }
+
+
+ /**
+ * Get the ring from the perspective of the instance.
+ */
+ public static List<RingInstanceDetails> ring(IInstance inst)
+ {
+ NodeToolResult results = inst.nodetoolResult("ring");
+ results.asserts().success();
+ return parseRing(results.getStdout());
+ }
+
+ /**
+ * Make sure the target instance is in the ring.
+ *
+ * @param instance instance to check on
+ * @param expectedInRing instance expected in the ring
+ * @return the ring (if target is present)
+ */
+ public static List<RingInstanceDetails> assertInRing(IInstance instance, IInstance expectedInRing)
+ {
+ String targetAddress = getBroadcastAddressHostString(expectedInRing);
+ List<RingInstanceDetails> ring = ring(instance);
+ Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+ assertThat(match).as("Not expected to find %s but was found", targetAddress).isPresent();
+ return ring;
+ }
+
+ /**
+ * Make sure the target instance's gossip state matches on the source instance
+ *
+ * @param instance instance to check on
+ * @param expectedInRing instance expected in the ring
+ * @param state expected gossip state
+ * @return the ring (if target is present and has expected state)
+ */
+ public static List<RingInstanceDetails> assertRingState(IInstance instance, IInstance expectedInRing, String state)
+ {
+ String targetAddress = getBroadcastAddressHostString(expectedInRing);
+ List<RingInstanceDetails> ring = ring(instance);
+ List<RingInstanceDetails> match = ring.stream()
+ .filter(d -> d.address.equals(targetAddress))
+ .collect(Collectors.toList());
+ assertThat(match)
+ .isNotEmpty()
+ .as("State was expected to be %s but was not", state)
+ .anyMatch(r -> r.state.equals(state));
+ return ring;
+ }
+
+ /**
+ * Make sure the target instance is NOT in the ring.
+ *
+ * @param instance instance to check on
+ * @param expectedInRing instance not expected in the ring
+ * @return the ring (if target is not present)
+ */
+ public static List<RingInstanceDetails> assertNotInRing(IInstance instance, IInstance expectedInRing)
+ {
+ String targetAddress = getBroadcastAddressHostString(expectedInRing);
+ List<RingInstanceDetails> ring = ring(instance);
+ Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(targetAddress)).findFirst();
+ Assert.assertEquals("Not expected to find " + targetAddress + " but was found", Optional.empty(), match);
+ return ring;
+ }
+
+ private static List<RingInstanceDetails> awaitRing(IInstance src, String errorMessage, Predicate<List<RingInstanceDetails>> fn)
+ {
+ List<RingInstanceDetails> ring = null;
+ for (int i = 0; i < 100; i++)
+ {
+ ring = ring(src);
+ if (fn.test(ring))
+ {
+ return ring;
+ }
+ sleepUninterruptibly(1, TimeUnit.SECONDS);
+ }
+ throw new AssertionError(errorMessage + "\n" + ring);
+ }
+
+ /**
+ * Wait for the target to be in the ring as seen by the source instance.
+ *
+ * @param instance instance to check on
+ * @param expectedInRing instance to wait for
+ * @return the ring
+ */
+ public static List<RingInstanceDetails> awaitRingJoin(IInstance instance, IInstance expectedInRing)
+ {
+ return awaitRingJoin(instance, expectedInRing.broadcastAddress().getAddress().getHostAddress());
+ }
+
+ /**
+ * Wait for the target to be in the ring as seen by the source instance.
+ *
+ * @param instance instance to check on
+ * @param expectedInRing instance address to wait for
+ * @return the ring
+ */
+ public static List<RingInstanceDetails> awaitRingJoin(IInstance instance, String expectedInRing)
+ {
+ return awaitRing(instance, "Node " + expectedInRing + " did not join the ring...", ring -> {
+ Optional<RingInstanceDetails> match = ring.stream().filter(d -> d.address.equals(expectedInRing)).findFirst();
+ if (match.isPresent())
+ {
+ RingInstanceDetails details = match.get();
+ return details.status.equals("Up") && details.state.equals("Normal");
+ }
+ return false;
+ });
+ }
+
+ /**
+ * Wait for the ring to only have instances that are Up and Normal.
+ *
+ * @param src instance to check on
+ * @return the ring
+ */
+ public static List<RingInstanceDetails> awaitRingHealthy(IInstance src)
+ {
+ return awaitRing(src, "Timeout waiting for ring to become healthy",
+ ring ->
+ ring.stream().allMatch(ClusterUtils::isRingInstanceDetailsHealthy));
+ }
+
+ /**
+ * Wait for the ring to have the target instance with the provided state.
+ *
+ * @param instance instance to check on
+ * @param expectedInRing to look for
+ * @param state expected
+ * @return the ring
+ */
+ public static List<RingInstanceDetails> awaitRingState(IInstance instance, IInstance expectedInRing, String state)
+ {
+ return awaitRing(instance, "Timeout waiting for " + expectedInRing + " to have state " + state,
+ ring ->
+ ring.stream()
+ .filter(d -> d.address.equals(getBroadcastAddressHostString(expectedInRing)))
+ .filter(d -> d.state.equals(state))
+ .findAny().isPresent());
+ }
+
+ /**
+ * Make sure the ring is only the expected instances. The source instance may not be in the ring, so this function
+ * only relies on the expectedInsts param.
+ *
+ * @param instance instance to check on
+ * @param expectedInRing expected instances in the ring
+ * @return the ring (if condition is true)
+ */
+ public static List<RingInstanceDetails> assertRingIs(IInstance instance, IInstance... expectedInRing)
+ {
+ return assertRingIs(instance, Arrays.asList(expectedInRing));
+ }
+
+ /**
+ * Make sure the ring is only the expected instances. The source instance may not be in the ring, so this function
+ * only relies on the expectedInsts param.
+ *
+ * @param instance instance to check on
+ * @param expectedInRing expected instances in the ring
+ * @return the ring (if condition is true)
+ */
+ public static List<RingInstanceDetails> assertRingIs(IInstance instance, Collection<? extends IInstance> expectedInRing)
+ {
+ Set<String> expectedRingAddresses = expectedInRing.stream()
+ .map(i -> i.config().broadcastAddress().getAddress().getHostAddress())
+ .collect(Collectors.toSet());
+ return assertRingIs(instance, expectedRingAddresses);
+ }
+
+ /**
+ * Make sure the ring is only the expected instances. The source instance may not be in the ring, so this function
+ * only relies on the expectedInsts param.
+ *
+ * @param instance instance to check on
+ * @param expectedRingAddresses expected instances addresses in the ring
+ * @return the ring (if condition is true)
+ */
+ public static List<RingInstanceDetails> assertRingIs(IInstance instance, Set<String> expectedRingAddresses)
+ {
+ List<RingInstanceDetails> ring = ring(instance);
+ Set<String> ringAddresses = ring.stream().map(d -> d.address).collect(Collectors.toSet());
+ assertThat(ringAddresses)
+ .as("Ring addreses did not match for instance %s", instance)
+ .isEqualTo(expectedRingAddresses);
+ return ring;
+ }
+
+ private static boolean isRingInstanceDetailsHealthy(RingInstanceDetails details)
+ {
+ return details.status.equals("Up") && details.state.equals("Normal");
+ }
+
+ private static List<RingInstanceDetails> parseRing(String str)
+ {
+ // 127.0.0.3 rack0 Up Normal 46.21 KB 100.00% -1
+ // /127.0.0.1:7012 Unknown ? Normal ? 100.00% -3074457345618258603
+ Pattern pattern = Pattern.compile("^(/?[0-9.:]+)\\s+(\\w+|\\?)\\s+(\\w+|\\?)\\s+(\\w+|\\?).*?(-?\\d+)\\s*$");
+ List<RingInstanceDetails> details = new ArrayList<>();
+ String[] lines = str.split("\n");
+ for (String line : lines)
+ {
+ Matcher matcher = pattern.matcher(line);
+ if (!matcher.find())
+ {
+ continue;
+ }
+ details.add(new RingInstanceDetails(matcher.group(1), matcher.group(2), matcher.group(3), matcher.group(4), matcher.group(5)));
+ }
+
+ return details;
+ }
+
+ private static Map<String, Map<String, String>> awaitGossip(IInstance src, String errorMessage, Predicate<Map<String, Map<String, String>>> fn)
+ {
+ Map<String, Map<String, String>> gossip = null;
+ for (int i = 0; i < 100; i++)
+ {
+ gossip = gossipInfo(src);
+ if (fn.test(gossip))
+ {
+ return gossip;
+ }
+ sleepUninterruptibly(1, TimeUnit.SECONDS);
+ }
+ throw new AssertionError(errorMessage + "\n" + gossip);
+ }
+
+ /**
+ * Wait for the target instance to have the desired status. Target status is checked via string contains so works
+ * with 'NORMAL' but also can check tokens or full state.
+ *
+ * @param instance instance to check on
+ * @param expectedInGossip instance to wait for
+ * @param targetStatus for the instance
+ * @return gossip info
+ */
+ public static Map<String, Map<String, String>> awaitGossipStatus(IInstance instance, IInstance expectedInGossip, String targetStatus)
+ {
+ return awaitGossip(instance, "Node " + expectedInGossip + " did not match state " + targetStatus, gossip -> {
+ Map<String, String> state = gossip.get(getBroadcastAddressString(expectedInGossip));
+ if (state == null)
+ return false;
+ String status = state.get("STATUS_WITH_PORT");
+ if (status == null)
+ status = state.get("STATUS");
+ if (status == null)
+ return targetStatus == null;
+ return status.contains(targetStatus);
+ });
+ }
+
+ /**
+ * Get the gossip information from the node. Currently only address, generation, and heartbeat are returned
+ *
+ * @param inst to check on
+ * @return gossip info
+ */
+ public static Map<String, Map<String, String>> gossipInfo(IInstance inst)
+ {
+ NodeToolResult results = inst.nodetoolResult("gossipinfo");
+ results.asserts().success();
+ return parseGossipInfo(results.getStdout());
+ }
+
+ /**
+ * Make sure the gossip info for the specific target has the expected generation and heartbeat
+ *
+ * @param instance to check on
+ * @param expectedInGossip instance to check for
+ * @param expectedGeneration expected generation
+ * @param expectedHeartbeat expected heartbeat
+ */
+ public static void assertGossipInfo(IInstance instance,
+ InetSocketAddress expectedInGossip, int expectedGeneration, int expectedHeartbeat)
+ {
+ String targetAddress = expectedInGossip.getAddress().toString();
+ Map<String, Map<String, String>> gossipInfo = gossipInfo(instance);
+ Map<String, String> gossipState = gossipInfo.get(targetAddress);
+ if (gossipState == null)
+ throw new NullPointerException("Unable to find gossip info for " + targetAddress + "; gossip info = " + gossipInfo);
+ Assert.assertEquals(Long.toString(expectedGeneration), gossipState.get("generation"));
+ Assert.assertEquals(Long.toString(expectedHeartbeat), gossipState.get("heartbeat")); //TODO do we really mix these two?
+ }
+
+ private static Map<String, Map<String, String>> parseGossipInfo(String str)
+ {
+ Map<String, Map<String, String>> map = new HashMap<>();
+ String[] lines = str.split("\n");
+ String currentInstance = null;
+ for (String line : lines)
+ {
+ if (line.startsWith("/"))
+ {
+ // start of new instance
+ currentInstance = line;
+ continue;
+ }
+ Objects.requireNonNull(currentInstance);
+ String[] kv = line.trim().split(":", 2);
+ assert kv.length == 2 : "When splitting line '" + line + "' expected 2 parts but not true";
+ Map<String, String> state = map.computeIfAbsent(currentInstance, ignore -> new HashMap<>());
+ state.put(kv[0], kv[1]);
+ }
+
+ return map;
+ }
+
+ /**
+ * Get the tokens assigned to the instance via config. This method does not work if the instance has learned
+ * or generated its tokens.
+ *
+ * @param instance to get tokens from
+ * @return non-empty list of tokens
+ */
+ public static List<String> getTokens(IInstance instance)
+ {
+ IInstanceConfig conf = instance.config();
+ int numTokens = conf.getInt("num_tokens");
+ Assert.assertEquals("Only single token is supported", 1, numTokens);
+ String token = conf.getString("initial_token");
+ Assert.assertNotNull("initial_token was not found", token);
+ return Arrays.asList(token);
+ }
+
+ /**
+ * Get all data directories for the given instance.
+ *
+ * @param instance to get data directories for
+ * @return data directories
+ */
+ public static List<File> getDataDirectories(IInstance instance)
+ {
+ IInstanceConfig conf = instance.config();
+ // this isn't safe as it assumes the implementation of InstanceConfig
+ // might need to get smarter... some day...
+ String[] ds = (String[]) conf.get("data_file_directories");
+ List<File> files = new ArrayList<>(ds.length);
+ for (int i = 0; i < ds.length; i++)
+ files.add(new File(ds[i]));
+ return files;
+ }
+
+ /**
+ * Get the commit log directory for the given instance.
+ *
+ * @param instance to get the commit log directory for
+ * @return commit log directory
+ */
+ public static File getCommitLogDirectory(IInstance instance)
+ {
+ IInstanceConfig conf = instance.config();
+ // this isn't safe as it assumes the implementation of InstanceConfig
+ // might need to get smarter... some day...
+ String d = (String) conf.get("commitlog_directory");
+ return new File(d);
+ }
+
+ /**
+ * Get the hints directory for the given instance.
+ *
+ * @param instance to get the hints directory for
+ * @return hints directory
+ */
+ public static File getHintsDirectory(IInstance instance)
+ {
+ IInstanceConfig conf = instance.config();
+ // this isn't safe as it assumes the implementation of InstanceConfig
+ // might need to get smarter... some day...
+ String d = (String) conf.get("hints_directory");
+ return new File(d);
+ }
+
+ /**
+ * Get the saved caches directory for the given instance.
+ *
+ * @param instance to get the saved caches directory for
+ * @return saved caches directory
+ */
+ public static File getSavedCachesDirectory(IInstance instance)
+ {
+ IInstanceConfig conf = instance.config();
+ // this isn't safe as it assumes the implementation of InstanceConfig
+ // might need to get smarter... some day...
+ String d = (String) conf.get("saved_caches_directory");
+ return new File(d);
+ }
+
+ /**
+ * Get all writable directories for the given instance.
+ *
+ * @param instance to get directories for
+ * @return all writable directories
+ */
+ public static List<File> getDirectories(IInstance instance)
+ {
+ List<File> out = new ArrayList<>();
+ out.addAll(getDataDirectories(instance));
+ out.add(getCommitLogDirectory(instance));
+ out.add(getHintsDirectory(instance));
+ out.add(getSavedCachesDirectory(instance));
+ return out;
+ }
+
+ /**
+ * Gets the name of the Partitioner for the given instance.
+ *
+ * @param instance to get partitioner from
+ * @return partitioner name
+ */
+ public static String getPartitionerName(IInstance instance)
+ {
+ return (String) instance.config().get("partitioner");
+ }
+
+ /**
+ * Changes the instance's address to the new address. This method should only be called while the instance is
+ * down, else has undefined behavior.
+ *
+ * @param instance to update address for
+ * @param address to set
+ */
+ public static void updateAddress(IInstance instance, String address)
+ {
+ updateAddress(instance.config(), address);
+ }
+
+ /**
+ * Changes the instance's address to the new address. This method should only be called while the instance is
+ * down, else has undefined behavior.
+ *
+ * @param conf to update address for
+ * @param address to set
+ */
+ private static void updateAddress(IInstanceConfig conf, String address)
+ {
+ for (String key : Arrays.asList("broadcast_address", "listen_address", "broadcast_rpc_address", "rpc_address"))
+ conf.set(key, address);
+
+ // InstanceConfig caches InetSocketAddress -> InetAddressAndPort
+ // this causes issues as startup now ignores config, so force reset it to pull from conf.
+ ((InstanceConfig) conf).unsetBroadcastAddressAndPort(); //TODO remove the need to null out the cache...
+ conf.networkTopology().put(conf.broadcastAddress(), NetworkTopology.dcAndRack(conf.localDatacenter(), conf.localRack()));
+ }
+
+ /**
+ * Get the broadcast address host address only (ex. 127.0.0.1)
+ */
+ private static String getBroadcastAddressHostString(IInstance target)
+ {
+ return target.config().broadcastAddress().getAddress().getHostAddress();
+ }
+
+ /**
+ * Get the broadcast address in host:port format (ex. 127.0.0.1:7190)
+ */
+ public static String getBroadcastAddressHostWithPortString(IInstance target)
+ {
+ InetSocketAddress address = target.config().broadcastAddress();
+ return address.getAddress().getHostAddress() + ":" + address.getPort();
+ }
+
+ /**
+ * Get the broadcast address InetAddess string (ex. localhost/127.0.0.1 or /127.0.0.1)
+ */
+ private static String getBroadcastAddressString(IInstance target)
+ {
+ return target.config().broadcastAddress().getAddress().toString();
+ }
+
+ public static final class RingInstanceDetails
+ {
+ private final String address;
+ private final String rack;
+ private final String status;
+ private final String state;
+ private final String token;
+
+ private RingInstanceDetails(String address, String rack, String status, String state, String token)
+ {
+ this.address = address;
+ this.rack = rack;
+ this.status = status;
+ this.state = state;
+ this.token = token;
+ }
+
+ public String getAddress()
+ {
+ return address;
+ }
+
+ public String getRack()
+ {
+ return rack;
+ }
+
+ public String getStatus()
+ {
+ return status;
+ }
+
+ public String getState()
+ {
+ return state;
+ }
+
+ public String getToken()
+ {
+ return token;
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ RingInstanceDetails that = (RingInstanceDetails) o;
+ return Objects.equals(address, that.address) &&
+ Objects.equals(rack, that.rack) &&
+ Objects.equals(status, that.status) &&
+ Objects.equals(state, that.state) &&
+ Objects.equals(token, that.token);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(address, rack, status, state, token);
+ }
+
+ public String toString()
+ {
+ return Arrays.asList(address, rack, status, state, token).toString();
+ }
+ }
+}
diff --cc test/distributed/org/apache/cassandra/distributed/test/BootstrapBinaryDisabledTest.java
index a7ac605caa,3ac5028557..55b6c86754
--- a/test/distributed/org/apache/cassandra/distributed/test/BootstrapBinaryDisabledTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/BootstrapBinaryDisabledTest.java
@@@ -36,7 -36,7 +36,7 @@@ import org.apache.cassandra.distributed
import org.apache.cassandra.distributed.api.TokenSupplier;
import org.apache.cassandra.distributed.shared.Byteman;
import org.apache.cassandra.distributed.shared.NetworkTopology;
--import org.apache.cassandra.distributed.shared.Shared;
++import org.apache.cassandra.utils.Shared;
/**
* Replaces python dtest bootstrap_test.py::TestBootstrap::test_bootstrap_binary_disabled
diff --cc test/distributed/org/apache/cassandra/distributed/test/ResourceLeakTest.java
index 5430800c02,177cbb9566..13fe696233
--- a/test/distributed/org/apache/cassandra/distributed/test/ResourceLeakTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/ResourceLeakTest.java
@@@ -33,9 -36,15 +36,12 @@@ import org.junit.Ignore
import org.junit.Test;
import com.sun.management.HotSpotDiagnosticMXBean;
-import org.apache.cassandra.db.Keyspace;
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.gms.Gossiper;
-import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.SigarLibrary;
import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
@@@ -148,7 -167,8 +161,8 @@@ public class ResourceLeakTest extends T
String tableName = "tbl" + loop;
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).callOnInstance(() -> FBUtilities.waitOnFutures(Keyspace.open(KEYSPACE).flush()));
+ cluster.get(1).flush(KEYSPACE);
+ actionToPerform.accept(cluster);
if (dumpEveryLoop)
{
dumpResources(String.format("loop%03d", loop));
diff --cc test/distributed/org/apache/cassandra/distributed/test/jmx/JMXFeatureTest.java
index 0000000000,1c38bd11e9..83a35e594d
mode 000000,100644..100644
--- a/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXFeatureTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXFeatureTest.java
@@@ -1,0 -1,86 +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.
- * NOTE: In later versions of Cassandra, there is also a `testOneNetworkInterfaceProvisioning` that leverages the ability to specify
- * ports in addition to IP/Host for binding, but this version does not support that feature. Keeping the test name the same
- * so that it's consistent across versions.
+ *
+ * @throws Exception
+ */
+ @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).withConfig(c -> c.with(Feature.values())).start())
++ 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()));
+ }
+ }
+ }
diff --cc test/distributed/org/apache/cassandra/distributed/test/jmx/JMXGetterCheckTest.java
index 0000000000,a3f74df60d..ece6052a75
mode 000000,100644..100644
--- a/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXGetterCheckTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXGetterCheckTest.java
@@@ -1,0 -1,138 +1,136 @@@
+ /*
+ * 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.util.ArrayList;
+ import java.util.Collections;
+ import java.util.List;
+ import java.util.Set;
+ import java.util.TreeSet;
+ import javax.management.JMRuntimeException;
+ import javax.management.MBeanAttributeInfo;
+ import javax.management.MBeanInfo;
+ import javax.management.MBeanOperationInfo;
+ import javax.management.MBeanServerConnection;
+ import javax.management.ObjectName;
+ import javax.management.remote.JMXConnector;
+ import javax.management.remote.JMXConnectorFactory;
+ import javax.management.remote.JMXServiceURL;
+
+ import com.google.common.collect.ImmutableSet;
+ import org.junit.Test;
+
+ import org.apache.cassandra.distributed.Cluster;
+ import org.apache.cassandra.distributed.api.Feature;
+ import org.apache.cassandra.distributed.api.IInvokableInstance;
+ import org.apache.cassandra.distributed.test.TestBaseImpl;
+
+ public class JMXGetterCheckTest extends TestBaseImpl
+ {
+ private static final Set<String> IGNORE_ATTRIBUTES = ImmutableSet.of(
+ "org.apache.cassandra.net:type=MessagingService:BackPressurePerHost" // throws unsupported saying the feature was removed... dropped in CASSANDRA-15375
+ );
+ private static final Set<String> IGNORE_OPERATIONS = ImmutableSet.of(
+ "org.apache.cassandra.db:type=StorageService:stopDaemon", // halts the instance, which then causes the JVM to exit
+ "org.apache.cassandra.db:type=StorageService:drain", // don't drain, it stops things which can cause other APIs to be unstable as we are in a stopped state
+ "org.apache.cassandra.db:type=StorageService:stopGossiping", // if we stop gossip this can cause other issues, so avoid
- "org.apache.cassandra.db:type=StorageService:resetLocalSchema", // this will fail when there are no other nodes which can serve schema
- "org.apache.cassandra.db:type=HintedHandoffManager:listEndpointsPendingHints", // this will fail because it only exists to match an old, deprecated mbean and just throws an UnsportedOperationException
- "org.apache.cassandra.db:type=StorageService:decommission" // Don't decommission nodes! Note that in future versions of C* this is unnecessary because decommission takes an argument.
++ "org.apache.cassandra.db:type=StorageService:resetLocalSchema" // this will fail when there are no other nodes which can serve schema
+ );
+
+ public static final String JMX_SERVICE_URL_FMT = "service:jmx:rmi:///jndi/rmi://%s:%d/jmxrmi";
+
+ @Test
+ public void testGetters() throws Exception
+ {
+ try (Cluster cluster = Cluster.build(1).withConfig(c -> c.with(Feature.values())).start())
+ {
+ 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))
+ {
+ MBeanServerConnection mbsc = jmxc.getMBeanServerConnection();
+ Set<ObjectName> metricNames = new TreeSet<>(mbsc.queryNames(null, null));
+ for (ObjectName name : metricNames)
+ {
+ if (!name.getDomain().startsWith("org.apache.cassandra"))
+ continue;
+ MBeanInfo info = mbsc.getMBeanInfo(name);
+ for (MBeanAttributeInfo a : info.getAttributes())
+ {
+ String fqn = String.format("%s:%s", name, a.getName());
+ if (!a.isReadable() || IGNORE_ATTRIBUTES.contains(fqn))
+ continue;
+ try
+ {
+ mbsc.getAttribute(name, a.getName());
+ }
+ catch (JMRuntimeException e)
+ {
+ errors.add(new Named(String.format("Attribute %s", fqn), e.getCause()));
+ }
+ }
+
+ for (MBeanOperationInfo o : info.getOperations())
+ {
+ String fqn = String.format("%s:%s", name, o.getName());
+ if (o.getSignature().length != 0 || IGNORE_OPERATIONS.contains(fqn))
+ continue;
+ try
+ {
+ mbsc.invoke(name, o.getName(), new Object[0], new String[0]);
+ }
+ catch (JMRuntimeException e)
+ {
+ errors.add(new Named(String.format("Operation %s", fqn), e.getCause()));
+ }
+ }
+ }
+ }
+ if (!errors.isEmpty())
+ {
+ AssertionError root = new AssertionError();
+ errors.forEach(root::addSuppressed);
+ throw root;
+ }
+ }
+ }
+
+ /**
+ * 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
+ {
+ public Named(String msg, Throwable cause)
+ {
+ super(msg + "\nCaused by: " + cause.getClass().getCanonicalName() + ": " + cause.getMessage(), cause.getCause());
+ StackTraceElement[] stack = cause.getStackTrace();
+ List<StackTraceElement> copy = new ArrayList<>();
+ for (StackTraceElement s : stack)
+ {
+ if (!s.getClassName().startsWith("org.apache.cassandra"))
+ break;
+ copy.add(s);
+ }
+ Collections.reverse(copy);
+ setStackTrace(copy.toArray(new StackTraceElement[0]));
+ }
+ }
+ }
diff --cc test/distributed/org/apache/cassandra/distributed/test/metric/TableMetricTest.java
index 3540acd302,0000000000..55121e5623
mode 100644,000000..100644
--- a/test/distributed/org/apache/cassandra/distributed/test/metric/TableMetricTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/metric/TableMetricTest.java
@@@ -1,271 -1,0 +1,288 @@@
+/*
+ * 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.metric;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+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;
+import org.junit.Test;
+
+import org.apache.cassandra.auth.AuthKeyspace;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.SystemKeyspace;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.repair.SystemDistributedKeyspace;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.tracing.TraceKeyspace;
+import org.apache.cassandra.utils.MBeanWrapper;
+
+import static org.apache.cassandra.config.CassandraRelevantProperties.IS_DISABLED_MBEAN_REGISTRATION;
+import static org.apache.cassandra.config.CassandraRelevantProperties.MBEAN_REGISTRATION_CLASS;
+
+public class TableMetricTest extends TestBaseImpl
+{
+ static
+ {
+ MBEAN_REGISTRATION_CLASS.setString(MapMBeanWrapper.class.getName());
+ IS_DISABLED_MBEAN_REGISTRATION.setBoolean(false);
+ }
-
+ private static volatile Map<String, Collection<String>> SYSTEM_TABLES = null;
+ private static Set<String> TABLE_METRIC_NAMES = ImmutableSet.of("WriteLatency");
+
+ /**
+ * Makes sure that all system tables have the expected metrics
+ * @throws IOException
+ */
+ @Test
+ public void systemTables() throws IOException
+ {
+ try (Cluster cluster = Cluster.build(1).start())
+ {
+ loadSystemTables(cluster);
+ assertSystemTableMetrics(cluster);
+ }
+ }
+
+ /**
+ * Tests that other table metrics are not modified when a single table is modified/deleted.
+ *
+ * @see <a href="https://issues.apache.org/jira/browse/CASSANDRA-16095">CASSANDRA-16095</a>
+ */
+ @Test
+ public void userTables() throws IOException
+ {
+ try (Cluster cluster = init(Cluster.build(3).start()))
+ {
+ loadSystemTables(cluster);
+ assertSystemTableMetrics(cluster);
+
+ cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (pk bigint PRIMARY KEY)"));
+ cluster.forEach(i -> assertTableMetricsExist(i, KEYSPACE, "tbl"));
+
+ // alter table can change metrics, so monitor for it
+ cluster.schemaChange(withKeyspace("ALTER TABLE %s.tbl WITH comment = 'testing'"));
+ cluster.forEach(i -> assertTableMetricsExist(i, KEYSPACE, "tbl"));
+
+ cluster.schemaChange(withKeyspace("ALTER TABLE %s.tbl ADD (value bigint)"));
+ cluster.forEach(i -> assertTableMetricsExist(i, KEYSPACE, "tbl"));
+
+ cluster.schemaChange(withKeyspace("ALTER TABLE %s.tbl RENAME pk TO pk2"));
+ cluster.forEach(i -> assertTableMetricsExist(i, KEYSPACE, "tbl"));
+
+ cluster.schemaChange(withKeyspace("ALTER TABLE %s.tbl DROP value"));
+ cluster.forEach(i -> assertTableMetricsExist(i, KEYSPACE, "tbl"));
+
+ // drop and make sure table no longer exists
+ cluster.schemaChange(withKeyspace("DROP TABLE %s.tbl"));
+ cluster.forEach(i -> assertTableMetricsDoesNotExist(i, KEYSPACE, "tbl"));
+
+ cluster.schemaChange(withKeyspace("DROP KEYSPACE %s"));
+ cluster.forEach(i -> assertKeyspaceMetricDoesNotExists(i, KEYSPACE));
+
+ // no other table impacted?
+ assertSystemTableMetrics(cluster);
+ }
+ }
+
+ private static void loadSystemTables(Cluster cluster)
+ {
+ SYSTEM_TABLES = cluster.get(1).callOnInstance(() -> {
+ Map<String, Collection<String>> map = new HashMap<>();
+ Arrays.asList(SystemKeyspace.metadata(), AuthKeyspace.metadata(), SystemDistributedKeyspace.metadata(),
+ Schema.getSystemKeyspaceMetadata(), TraceKeyspace.metadata())
+ .forEach(meta -> {
+ Set<String> tables = meta.tables.stream().map(t -> t.name).collect(Collectors.toSet());
+ map.put(meta.name, tables);
+ });
+ return map;
+ });
+ }
+
+ private static void assertSystemTableMetrics(Cluster cluster)
+ {
+ for (String keyspace : SYSTEM_TABLES.keySet())
+ {
+ for (String table : SYSTEM_TABLES.get(keyspace))
+ {
+ cluster.forEach(i -> assertTableMetricsExist(i, keyspace, table));
+ }
+ }
+ }
+
+ private static void assertTableMetricsExist(IInvokableInstance inst, String keyspace, String table)
+ {
+ assertTableMBeanExists(inst, keyspace, table);
+ for (String metric : TABLE_METRIC_NAMES)
+ assertTableMetricExists(inst, keyspace, table, metric);
+ }
+
+ private static void assertTableMetricsDoesNotExist(IInvokableInstance inst, String keyspace, String table)
+ {
+ assertTableMBeanDoesNotExists(inst, keyspace, table);
+ for (String metric : TABLE_METRIC_NAMES)
+ assertTableMetricDoesNotExists(inst, keyspace, table, metric);
+ }
+
+ private static void assertKeyspaceMetricDoesNotExists(IInvokableInstance inst, String keyspace)
+ {
+ for (String metric : TABLE_METRIC_NAMES)
+ assertKeyspaceMetricDoesNotExists(inst, keyspace, metric);
+ }
+
+ private static void assertTableMBeanExists(IInvokableInstance inst, String keyspace, String table)
+ {
+ 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,
+ mbeans.isRegistered(ColumnFamilyStore.getColumnFamilieMBeanName(keyspace, table, false)));
+ });
+ }
+
+ private static void assertTableMBeanDoesNotExists(IInvokableInstance inst, String keyspace, String table)
+ {
+ 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,
+ mbeans.isRegistered(ColumnFamilyStore.getColumnFamilieMBeanName(keyspace, table, false)));
+ });
+ }
+
+ private static void assertTableMetricExists(IInvokableInstance inst, String keyspace, String table, String name)
+ {
+ 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));
+
+ // verify replicated to keyspace
+ String keyspaceMBean = getKeyspaceMetricName(keyspace, name);
+ Assert.assertTrue("Unable to find keyspace metric " + keyspaceMBean + " for " + keyspace, mbeans.isRegistered(keyspaceMBean));
+ });
+ }
+
+ private static void assertTableMetricDoesNotExists(IInvokableInstance inst, String keyspace, String table, String name)
+ {
+ 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));
+
+ // validate keyspace metric
+ assertKeyspaceMetricMayExists(mbeans, keyspace, name);
+ });
+ }
+
+ private static void assertKeyspaceMetricMayExists(MapMBeanWrapper mbeans, String keyspace, String name)
+ {
+ String keyspaceMBean = getKeyspaceMetricName(keyspace, name);
+ boolean keyspaceExists = Schema.instance.getKeyspaceMetadata(keyspace) != null;
+ String errorMessage = keyspaceExists ?
+ "Unable to find keyspace metric " + keyspaceMBean + " for " + keyspace :
+ "Found keyspace metric " + keyspaceMBean + " for " + keyspace;
+ Assert.assertEquals(errorMessage, keyspaceExists, mbeans.isRegistered(keyspaceMBean));
+ }
+
+ private static void assertKeyspaceMetricDoesNotExists(IInvokableInstance inst, String keyspace, String name)
+ {
+ 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));
+ });
+ }
+
+ private static String getKeyspaceMetricName(String keyspace, String name)
+ {
+ return String.format("org.apache.cassandra.metrics:type=Keyspace,keyspace=%s,name=%s", keyspace, name);
+ }
+
+ private static String getTableMetricName(String keyspace, String table, String name)
+ {
+ 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<>();
+
+ @Override
+ public void registerMBean(Object obj, ObjectName mbeanName, OnException onException)
+ {
+ Object current = map.putIfAbsent(mbeanName, obj);
+ if (current != null)
+ onException.handler.accept(new InstanceAlreadyExistsException("MBean " + mbeanName + " already exists"));
+ }
+
+ @Override
+ public boolean isRegistered(ObjectName mbeanName, OnException onException)
+ {
+ return map.containsKey(mbeanName);
+ }
+
+ @Override
+ public void unregisterMBean(ObjectName mbeanName, OnException onException)
+ {
+ Object previous = map.remove(mbeanName);
+ 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;
++ }
+ }
+}
diff --cc test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeMessageForwardTest.java
index 29c3209a6f,0000000000..23dc2a9a08
mode 100644,000000..100644
--- a/test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeMessageForwardTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeMessageForwardTest.java
@@@ -1,125 -1,0 +1,125 @@@
+/*
+ * 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.upgrade;
+
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.distributed.UpgradeableCluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.IUpgradeableInstance;
- import org.apache.cassandra.distributed.shared.Shared;
++import org.apache.cassandra.utils.Shared;
+import org.awaitility.Awaitility;
+
+import static org.apache.cassandra.distributed.shared.AssertUtils.*;
+
+@Shared
+public class MixedModeMessageForwardTest extends UpgradeTestBase
+{
+ private static final Logger logger = LoggerFactory.getLogger(MixedModeMessageForwardTest.class);
+ private static int nextKey = 1;
+ private static final String TABLE = "tbl";
+ private static final String INSERT_QUERY = String.format("INSERT INTO %s.%s(pk) VALUES (?)", KEYSPACE, TABLE);
+ private static final String CHECK_QUERY = String.format("SELECT pk FROM %s.%s WHERE pk = ?", KEYSPACE, TABLE);
+
+ static boolean checkSelectQueriesRespond(IUpgradeableInstance instance, int keyToInsert)
+ {
+ try
+ {
+ instance.coordinator().execute(CHECK_QUERY, ConsistencyLevel.ALL, 0);
+ return true;
+ }
+ catch (Throwable tr)
+ {
+ logger.info("Select CL.ALL failed, retrying.", tr);
+ return false;
+ }
+ }
+
+ private void writeReadTest(UpgradeableCluster cluster)
+ {
+ // Coordinate a write from each node and then check present on all replicas
+ for (int coordId = 1; coordId <= cluster.size(); coordId++)
+ {
+ final IUpgradeableInstance instance = cluster.get(coordId);
+ final int keyToInsert = nextKey++;
+
+ // Wait for the messaging service to be connected for up to a minute by issuing
+ // a CL.ALL read that requires a connection to all other instances
+ Awaitility.await("MessagingService ready CL.ALL select from node" + coordId)
+ .atMost(1, TimeUnit.MINUTES)
+ .until(() -> checkSelectQueriesRespond(instance, keyToInsert));
+
+ cluster.get(coordId).coordinator().execute(INSERT_QUERY, ConsistencyLevel.ALL, keyToInsert);
+
+ for (int nodeId = 1; nodeId <= cluster.size(); nodeId++) {
+ Object[][] results = cluster.get(nodeId).executeInternal(CHECK_QUERY, keyToInsert);
+ assertRows(results, row(keyToInsert));
+ }
+ }
+ }
+
+ /* Verify that messages sent with sendToHintedReplicas to non-local DCs
+ * are forwarded on to the hosts there.
+ *
+ * 1) creates a mixed cluster with multiple datacenters and a keyspace
+ * configured to write to all replicas in the datacenter
+ * 2) check the original single-version cluster by issuing an INSERT
+ * mutation from a coordinator on each node, then check that value
+ * has locally been written to each of the nodes. INSERT happens until ALL nodes
+ * respond to give time for internode messaging to be established after upgrade.
+ * 3) Upgrade nodes one at a time, rechecking that all writes are forwarded.
+ */
+ @Test
+ public void checkWritesForwardedToOtherDcTest() throws Throwable
+ {
+ int numDCs = 2;
+ int nodesPerDc = 2;
+ String ntsArgs = IntStream.range(1, numDCs + 1)
+ .mapToObj(dc -> String.format("'datacenter%d' : %d", dc, nodesPerDc))
+ .collect(Collectors.joining(","));
+
+ new TestCase()
+ .withConfig(c -> c.with(Feature.GOSSIP, Feature.NETWORK))
+ .withBuilder(b -> b.withRacks(numDCs, 1, nodesPerDc))
+ .nodes(numDCs * nodesPerDc)
+ .singleUpgrade(v30, v40)
+ .setup(cluster -> {
+ cluster.schemaChange("ALTER KEYSPACE " + KEYSPACE +
+ " WITH replication = {'class': 'NetworkTopologyStrategy', " + ntsArgs + " };");
+
+ cluster.schemaChange(String.format("CREATE TABLE %s.%s (pk int, PRIMARY KEY(pk))", KEYSPACE, TABLE));
+
+ logger.info("Testing after setup, all nodes running {}", cluster.get(1).getReleaseVersionString());
+ writeReadTest(cluster);
+ })
+ .runAfterNodeUpgrade((UpgradeableCluster cluster, int nodeId) -> {
+ // Should be able to coordinate a write to any node and have a copy appear locally on all others
+ logger.info("Testing after upgrading node{} to {}", nodeId, cluster.get(nodeId).getReleaseVersionString());
+ writeReadTest(cluster);
+ })
+ .run();
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org