diff --git a/.build/parent-pom-template.xml b/.build/parent-pom-template.xml
index b6f4b63e229f..b874c7a857c1 100644
--- a/.build/parent-pom-template.xml
+++ b/.build/parent-pom-template.xml
@@ -36,8 +36,8 @@
     </license>
   </licenses>
   <properties>
-    <bytebuddy.version>1.12.13</bytebuddy.version>
-    <byteman.version>4.0.20</byteman.version>
+    <bytebuddy.version>1.14.17</bytebuddy.version>
+    <byteman.version>4.0.23</byteman.version>
     <ohc.version>0.5.1</ohc.version>
 
     <!-- These are referenced in build.xml, so need to be propagated from there -->
@@ -486,13 +486,13 @@
       <dependency>
         <groupId>org.mockito</groupId>
         <artifactId>mockito-core</artifactId>
-        <version>4.7.0</version>
+        <version>5.12.0</version>
         <scope>test</scope>
       </dependency>
       <dependency>
         <groupId>org.mockito</groupId>
         <artifactId>mockito-inline</artifactId>
-        <version>4.7.0</version>
+        <version>5.2.0</version>
         <scope>test</scope>
       </dependency>
       <dependency>
@@ -734,7 +734,7 @@
       <dependency>
         <groupId>io.netty</groupId>
         <artifactId>netty-all</artifactId>
-        <version>4.1.96.Final</version>
+        <version>4.1.110.Final</version>
         <exclusions>
           <exclusion>
             <groupId>io.netty</groupId>
@@ -805,28 +805,28 @@
       <dependency>
         <groupId>io.netty</groupId>
         <artifactId>netty-tcnative-boringssl-static</artifactId>
-        <version>2.0.61.Final</version>
+        <version>2.0.65.Final</version>
       </dependency>
       <dependency>
         <groupId>org.bouncycastle</groupId>
         <artifactId>bcpkix-jdk18on</artifactId>
-        <version>1.76</version>
+        <version>1.78.1</version>
       </dependency>
       <dependency>
         <groupId>io.netty</groupId>
         <artifactId>netty-transport-native-epoll</artifactId>
-        <version>4.1.96.Final</version>
+        <version>4.1.110.Final</version>
       </dependency>
       <dependency>
         <groupId>io.netty</groupId>
         <artifactId>netty-transport-native-epoll</artifactId>
-        <version>4.1.96.Final</version>
+        <version>4.1.110.Final</version>
         <classifier>linux-x86_64</classifier>
       </dependency>
       <dependency>
         <groupId>io.netty</groupId>
         <artifactId>netty-transport-native-epoll</artifactId>
-        <version>4.1.96.Final</version>
+        <version>4.1.110.Final</version>
         <classifier>linux-aarch_64</classifier>
       </dependency>
 
@@ -1065,7 +1065,7 @@
       <dependency>
         <groupId>org.jctools</groupId>
         <artifactId>jctools-core</artifactId>
-        <version>3.1.0</version>
+        <version>3.3.0</version>
       </dependency>
 
       <dependency>
diff --git a/bin/cassandra.in.sh b/bin/cassandra.in.sh
index 237f733ecf48..f6cb2e2efc2c 100644
--- a/bin/cassandra.in.sh
+++ b/bin/cassandra.in.sh
@@ -122,11 +122,13 @@ jvmver=`echo "$java_ver_output" | grep '[openjdk|java] version' | awk -F'"' 'NR=
 JVM_VERSION=${jvmver%_*}
 short=$(echo "${jvmver}" | cut -c1-2)
 
-JAVA_VERSION=17
+JAVA_VERSION=22
 if [ "$short" = "11" ]  ; then
      JAVA_VERSION=11
-elif [ "$JVM_VERSION" \< "17" ] ; then
-    echo "DSE DB 5.0 requires Java 11 or Java 17."
+elif [ "$short" = "17" ]  ; then
+     JAVA_VERSION=17
+elif [ "$JVM_VERSION" \< "22" ] ; then
+    echo "DSE DB 5.0 requires Java 11, Java 17 or Java 22."
     exit 1;
 fi
 
@@ -149,9 +151,11 @@ case "$jvm" in
         ;;
 esac
 
-# Read user-defined JVM options from jvm-server.options file
+# Read user-defined JVM options from jvm-clients.options file
 JVM_OPTS_FILE=$CASSANDRA_CONF/jvm${jvmoptions_variant:--clients}.options
-if [ $JAVA_VERSION -ge 17 ] ; then
+if [ $JAVA_VERSION -ge 22 ] ; then
+    JVM_DEP_OPTS_FILE=$CASSANDRA_CONF/jvm22${jvmoptions_variant:--clients}.options
+elif [ $JAVA_VERSION -ge 17 ] ; then
     JVM_DEP_OPTS_FILE=$CASSANDRA_CONF/jvm17${jvmoptions_variant:--clients}.options
 elif [ $JAVA_VERSION -ge 11 ] ; then
     JVM_DEP_OPTS_FILE=$CASSANDRA_CONF/jvm11${jvmoptions_variant:--clients}.options
diff --git a/build.xml b/build.xml
index 3f9fecb6c77a..069d42a9327d 100644
--- a/build.xml
+++ b/build.xml
@@ -45,7 +45,7 @@
         The use of both CASSANDRA_USE_JDK11 and use-jdk11 is deprecated.
     -->
     <property name="java.default" value="11" />
-    <property name="java.supported" value="11,17" />
+    <property name="java.supported" value="11,17,22" />
 
     <!-- directory details -->
     <property name="basedir" value="."/>
@@ -308,6 +308,7 @@
         <string>--add-opens java.base/java.lang=ALL-UNNAMED</string>
         <string>--add-opens java.base/java.util=ALL-UNNAMED</string>
         <string>--add-opens java.base/java.nio=ALL-UNNAMED</string>
+        <string>--add-opens java.base/java.nio.file.attribute=ALL-UNNAMED</string>
 
         <string>--add-opens java.rmi/sun.rmi.transport.tcp=ALL-UNNAMED</string>
     </resources>
@@ -316,6 +317,77 @@
         <equals arg1="${ant.java.version}" arg2="17"/>
     </condition>
 
+    <resources id="_jvm22_arg_items">
+        <string>-XX:+UnlockDiagnosticVMOptions</string>
+        <string>-Djdk.attach.allowAttachSelf=true</string>
+        <string>-XX:+UseG1GC</string>
+        <string>-XX:+ParallelRefProcEnabled</string>
+        <!-- Below two are used in Cassandra, not in CNDB-->
+        <!--<string>-XX:MaxTenuringThreshold=1</string>
+        <string>-XX:G1HeapRegionSize=16m</string>-->
+        <string>-XX:G1RSetUpdatingPauseTimePercent=5</string>
+        <string>-XX:MaxGCPauseMillis=100</string>
+
+        <!-- Copy-paste for now these options from cndb/docker-entrypoint.sh; we will verify later whether we
+         want all of them for tests-->
+        <string>-XX:-RestrictContended</string>
+        <string>-XX:+UseThreadPriorities</string>
+        <string>-XX:+DebugNonSafepoints</string>
+        <string>-XX:+UseStringDeduplication</string>
+        <string>-XX:StringTableSize=1000003</string>
+        <string>-XX:+PerfDisableSharedMem</string>
+        <string>-XX:+AlwaysPreTouch</string>
+        <string>-XX:+UseTLAB</string>
+        <string>-XX:+ResizeTLAB</string>
+        <string>-XX:+UseNUMA</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/jdk.internal.perf=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 jdk.naming.dns/com.sun.jndi.dns=java.naming</string>
+        <string>--add-exports jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED</string>
+        <string>--add-exports jdk.unsupported/sun.misc=ALL-UNNAMED</string>
+
+        <string>--add-opens java.base/jdk.internal.module=ALL-UNNAMED</string>
+        <string>--add-opens java.base/jdk.internal.reflect=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.math=ALL-UNNAMED</string>
+        <string>--add-opens java.base/java.math=ALL-UNNAMED</string>
+        <string>--add-opens jdk.management.jfr/jdk.management.jfr=ALL-UNNAMED</string>
+        <string>--add-opens jdk.management/com.sun.management.internal=ALL-UNNAMED</string>
+        <string>--add-opens jdk.naming.dns/com.sun.jndi.dns=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/java.io=ALL-UNNAMED</string>
+        <string>--add-opens java.base/sun.nio.ch=ALL-UNNAMED</string>
+        <string>--add-opens java.base/java.nio=ALL-UNNAMED</string>
+        <string>--add-opens java.base/java.nio.file.spi=ALL-UNNAMED</string>
+        <string>--add-opens java.base/java.lang=ALL-UNNAMED</string>
+        <string>--add-opens java.base/java.util=ALL-UNNAMED</string>
+        <string>--add-opens java.base/java.lang.reflect=ALL-UNNAMED</string>
+        <string>--add-opens jdk.compiler/com.sun.tools.javac=ALL-UNNAMED</string>
+
+        <!-- required for org.apache.cassandra.Util.getSupportedMTimeGranularity-->
+        <string>--add-opens java.base/java.nio.file.attribute=ALL-UNNAMED</string>
+
+        <!-- below was added in Cassandra only; check for what and whether we need it here
+            add-opens java.rmi/sun.rmi.transport.tcp=ALL-UNNAMED-->
+
+        <!-- below 2 were added to support jamm and fix BatchCommitLogTest -->
+        <string>--add-opens java.base/java.util.concurrent=ALL-UNNAMED</string>
+        <string>--add-opens java.base/java.util.concurrent.atomic=ALL-UNNAMED</string>
+    </resources>
+    <pathconvert property="_jvm22_args_concat" refid="_jvm22_arg_items" pathsep=" "/>
+    <condition property="java-jvmargs" value="${_jvm22_args_concat}">
+    <equals arg1="${ant.java.version}" arg2="22"/>
+    </condition>
+
     <!--
       JVM arguments for tests.
 
@@ -343,6 +415,19 @@
     <condition property="_std-test-jvmargs" value="${_jvm17_test_arg_items_concat}">
         <equals arg1="${ant.java.version}" arg2="17"/>
     </condition>
+    <resources id="_jvm22_test_arg_items">
+    <string>-Djava.security.manager=allow</string>
+    <string>-Dio.netty.tryReflectionSetAccessible=true</string>
+        <!-- Enable vector incubator feature (simd support)-->
+        <string>--add-modules jdk.incubator.vector</string>
+    </resources>
+    <pathconvert property="_jvm22_test_arg_items_concat" refid="_jvm22_test_arg_items" pathsep=" "/>
+    <condition property="_std-test-jvmargs" value="${_jvm11_test_arg_items_concat}">
+        <equals arg1="${ant.java.version}" arg2="11"/>
+    </condition>
+    <condition property="_std-test-jvmargs" value="${_jvm22_test_arg_items_concat}">
+    <equals arg1="${ant.java.version}" arg2="22"/>
+    </condition>
 
     <!-- needed to compile org.apache.cassandra.utils.JMXServerUtils -->
     <!-- needed to compile org.apache.cassandra.distributed.impl.Instance-->
@@ -1250,6 +1335,7 @@
       <jvmarg value="-Dcassandra.ring_delay_ms=1000"/>
       <jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
       <jvmarg value="-Dcassandra.skip_sync=true" />
+      <jvmarg value="-Dnet.bytebuddy.experimental=true"/>
     </testmacro>
     <fileset dir="${test.unit.src}" />
   </target>
@@ -1267,6 +1353,7 @@
         <jvmarg value="-Dcassandra.ring_delay_ms=1000"/>
         <jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
         <jvmarg value="-Dcassandra.skip_sync=true" />
+        <jvmarg value="-Dnet.bytebuddy.experimental=true"/>
         <jvmarg value="-Xmx${maxMemory}"/>
         <jvmarg value="-Dlogback.configurationFile=file://${test.logback.configurationFile}"/>
       </testmacrohelper>
@@ -1292,6 +1379,7 @@
         <jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
         <jvmarg value="-Dcassandra.config=file:///${compressed_yaml}"/>
         <jvmarg value="-Dcassandra.skip_sync=true" />
+        <jvmarg value="-Dnet.bytebuddy.experimental=true"/>
         <jvmarg value="-Dlogback.configurationFile=file://${test.logback.configurationFile}"/>
       </testmacrohelper>
     </sequential>
@@ -1313,6 +1401,7 @@
         <jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
         <jvmarg value="-Dcassandra.config=file:///${cdc_yaml}"/>
         <jvmarg value="-Dcassandra.skip_sync=true" />
+        <jvmarg value="-Dnet.bytebuddy.experimental=true"/>
         <jvmarg value="-Dlogback.configurationFile=file://${test.logback.configurationFile}"/>
       </testmacrohelper>
     </sequential>
@@ -1336,6 +1425,7 @@
                 <jvmarg value="-Dcassandra.config=file:///${latest_yaml}"/>
                 <jvmarg value="-Dcassandra.test.storage_compatibility_mode=NONE"/>
                 <jvmarg value="-Dcassandra.skip_sync=true" />
+                <jvmarg value="-Dnet.bytebuddy.experimental=true"/>
                 <jvmarg value="-Dlogback.configurationFile=file://${test.logback.configurationFile}"/>
             </testmacrohelper>
         </sequential>
@@ -1358,6 +1448,7 @@
         <jvmarg value="-Dcassandra.config=file:///${scm_none_yaml}"/>
         <jvmarg value="-Dcassandra.test.storage_compatibility_mode=NONE"/>
         <jvmarg value="-Dcassandra.skip_sync=true" />
+        <jvmarg value="-Dnet.bytebuddy.experimental=true"/>
         <jvmarg value="-Dlogback.configurationFile=file://${test.logback.configurationFile}"/>
       </testmacrohelper>
     </sequential>
@@ -1375,6 +1466,7 @@
                 <jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
                 <jvmarg value="-Dcassandra.config.loader=org.apache.cassandra.OffsetAwareConfigurationLoader"/>
                 <jvmarg value="-Dcassandra.skip_sync=true" />
+                <jvmarg value="-Dnet.bytebuddy.experimental=true"/>
                 <jvmarg value="-Dcassandra.allow_cursor_compaction=false" />
                 <jvmarg value="-Dlogback.configurationFile=file://${test.logback.configurationFile}"/>
             </testmacrohelper>
@@ -1397,6 +1489,7 @@
         <jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
         <jvmarg value="-Dcassandra.config=file:///${system_keyspaces_directory_yaml}"/>
         <jvmarg value="-Dcassandra.skip_sync=true" />
+        <jvmarg value="-Dnet.bytebuddy.experimental=true"/>
         <jvmarg value="-Dlogback.configurationFile=file://${test.logback.configurationFile}"/>
       </testmacrohelper>
     </sequential>
@@ -1413,6 +1506,7 @@
             <jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
             <jvmarg value="-Dcassandra.sai.latest.version=aa"/>
             <jvmarg value="-Dcassandra.skip_sync=true" />
+            <jvmarg value="-Dnet.bytebuddy.experimental=true"/>
             <jvmarg value="-Dlogback.configurationFile=file://${test.logback.configurationFile}"/>
         </testmacrohelper>
     </sequential>
@@ -1428,6 +1522,7 @@
               <jvmarg value="-Dcassandra.ring_delay_ms=1000"/>
               <jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
               <jvmarg value="-Dcassandra.skip_sync=true" />
+              <jvmarg value="-Dnet.bytebuddy.experimental=true"/>
               <jvmarg value="-Dlogback.configurationFile=file://\${test.logback.configurationFile}"/>
           </testmacrohelper>
       </sequential>
@@ -1458,6 +1553,7 @@
       <jvmarg value="-Dcassandra.ring_delay_ms=1000"/>
       <jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
       <jvmarg value="-Dcassandra.skip_sync=true" />
+      <jvmarg value="-Dnet.bytebuddy.experimental=true"/>
     </testmacro>
   </target>
 
@@ -1621,6 +1717,7 @@
         <jvmarg value="-Dcassandra.memtable_row_overhead_computation_step=100"/>
         <jvmarg value="-Dcassandra.test.use_prepared=${cassandra.test.use_prepared}"/>
         <jvmarg value="-Dcassandra.skip_sync=true" />
+        <jvmarg value="-Dnet.bytebuddy.experimental=true"/>
         <classpath>
           <path refid="cassandra.classpath.test" />
           <pathelement location="${test.classes}"/>
@@ -1667,6 +1764,7 @@
         <jvmarg value="-Dcassandra.test.use_prepared=${cassandra.test.use_prepared}"/>
         <jvmarg value="-Dcassandra.memtable_row_overhead_computation_step=100"/>
         <jvmarg value="-Dcassandra.skip_sync=true" />
+        <jvmarg value="-Dnet.bytebuddy.experimental=true"/>
         <classpath>
           <path refid="cassandra.classpath.test" />
           <pathelement location="${test.classes}"/>
@@ -1890,6 +1988,7 @@
       <jvmarg value="-Dcassandra.ring_delay_ms=10000"/>
       <jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
       <jvmarg value="-Dcassandra.skip_sync=true" />
+      <jvmarg value="-Dnet.bytebuddy.experimental=true"/>
       <jvmarg value="-Xmx8G"/>
     </testmacro>
   </target>
@@ -1901,6 +2000,7 @@
       <jvmarg value="-Dcassandra.ring_delay_ms=10000"/>
       <jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
       <jvmarg value="-Dcassandra.skip_sync=true" />
+      <jvmarg value="-Dnet.bytebuddy.experimental=true"/>
     </testmacro>
   </target>
 
@@ -1911,6 +2011,7 @@
       <jvmarg value="-Dcassandra.ring_delay_ms=10000"/>
       <jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
       <jvmarg value="-Dcassandra.skip_sync=true" />
+      <jvmarg value="-Dnet.bytebuddy.experimental=true"/>
       <jvmarg value="-Dcassandra.debugrefcount=false"/>
       <jvmarg value="-Dcassandra.test.simulator.determinismcheck=strict"/>
       <jvmarg value="-Dcassandra.test.simulator.print_asm=${simulator.asm.print}" />
@@ -1937,6 +2038,7 @@
       <jvmarg value="-Dcassandra.ring_delay_ms=10000"/>
       <jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
       <jvmarg value="-Dcassandra.skip_sync=true" />
+      <jvmarg value="-Dnet.bytebuddy.experimental=true"/>
       <jvmarg value="-Xmx8G"/>
     </testmacro>
   </target>
@@ -1953,6 +2055,7 @@
       <jvmarg value="-Dcassandra.ring_delay_ms=10000"/>
       <jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
       <jvmarg value="-Dcassandra.skip_sync=true" />
+      <jvmarg value="-Dnet.bytebuddy.experimental=true"/>
       <jvmarg value="-Xmx8G"/>
     </testmacro>
   </target>
@@ -1966,6 +2069,7 @@
       <jvmarg value="-Dcassandra.ring_delay_ms=10000"/>
       <jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
       <jvmarg value="-Dcassandra.skip_sync=true" />
+      <jvmarg value="-Dnet.bytebuddy.experimental=true"/>
     </testmacro>
   </target>
 
diff --git a/conf/jvm-server.options b/conf/jvm-server.options
index c50f3631f5a0..94b50c89bd0a 100644
--- a/conf/jvm-server.options
+++ b/conf/jvm-server.options
@@ -171,6 +171,9 @@
 #-Xms4G
 #-Xmx4G
 
+# Need experimental bytebuddy for JDK21
+-Dnet.bytebuddy.experimental
+
 # Young generation size is automatically calculated by cassandra-env
 # based on this formula: min(100 * num_cores, 1/4 * heap size)
 #
diff --git a/conf/jvm17-server.options b/conf/jvm17-server.options
index e054847ed640..23ddb84a6415 100644
--- a/conf/jvm17-server.options
+++ b/conf/jvm17-server.options
@@ -97,7 +97,7 @@
 # Java 11 (and newer) GC logging options:
 # See description of https://bugs.openjdk.java.net/browse/JDK-8046148 for details about the syntax
 # The following is the equivalent to -XX:+PrintGCDetails -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10 -XX:GCLogFileSize=10M
--Xlog:gc=info,heap*=debug,age*=debug,safepoint=info,promotion*=debug:file=/var/log/cassandra/gc.log:time,uptime,pid,tid,level:filecount=10,filesize=10485760
+#-Xlog:gc=info,heap*=debug,age*=debug,safepoint=info,promotion*=debug:file=/var/log/cassandra/gc.log:time,uptime,pid,tid,level:filecount=10,filesize=10485760
 
 # Notes for Java 8 migration:
 #
diff --git a/conf/jvm22-clients.options b/conf/jvm22-clients.options
new file mode 100644
index 000000000000..6e30351a77ee
--- /dev/null
+++ b/conf/jvm22-clients.options
@@ -0,0 +1,31 @@
+###########################################################################
+#                         jvm22-clients.options                           #
+#                                                                         #
+# See jvm-clients.options. This file is specific for Java 22 and newer.   #
+###########################################################################
+
+###################
+#  JPMS SETTINGS  #
+###################
+
+-Djdk.attach.allowAttachSelf=true
+--add-exports java.base/jdk.internal.misc=ALL-UNNAMED
+--add-exports java.base/jdk.internal.ref=ALL-UNNAMED
+--add-exports java.base/sun.nio.ch=ALL-UNNAMED
+--add-exports java.management.rmi/com.sun.jmx.remote.internal.rmi=ALL-UNNAMED
+--add-exports java.rmi/sun.rmi.registry=ALL-UNNAMED
+--add-exports java.rmi/sun.rmi.server=ALL-UNNAMED
+--add-exports java.sql/java.sql=ALL-UNNAMED
+
+--add-opens java.base/java.lang.module=ALL-UNNAMED
+--add-opens java.base/jdk.internal.loader=ALL-UNNAMED
+--add-opens java.base/jdk.internal.ref=ALL-UNNAMED
+--add-opens java.base/jdk.internal.reflect=ALL-UNNAMED
+--add-opens java.base/jdk.internal.math=ALL-UNNAMED
+--add-opens java.base/jdk.internal.module=ALL-UNNAMED
+--add-opens jdk.management/com.sun.management.internal=ALL-UNNAMED
+--add-opens=java.base/java.lang.reflect=ALL-UNNAMED
+--add-opens java.base/java.io=ALL-UNNAMED
+--add-opens java.base/sun.nio.ch=ALL-UNNAMED
+
+# The newline in the end of file is intentional
\ No newline at end of file
diff --git a/conf/jvm22-server.options b/conf/jvm22-server.options
new file mode 100644
index 000000000000..b46df5fc5fe3
--- /dev/null
+++ b/conf/jvm22-server.options
@@ -0,0 +1,126 @@
+###########################################################################
+#                         jvm22-server.options                            #
+#                                                                         #
+# See jvm-server.options. This file is specific for Java 22 and newer.    #
+###########################################################################
+
+#################
+#  GC SETTINGS  #
+#################
+
+
+
+### CMS Settings
+#-XX:+UseConcMarkSweepGC
+#-XX:+CMSParallelRemarkEnabled
+#-XX:SurvivorRatio=8
+#-XX:MaxTenuringThreshold=1
+#-XX:CMSInitiatingOccupancyFraction=75
+#-XX:+UseCMSInitiatingOccupancyOnly
+#-XX:CMSWaitDuration=10000
+#-XX:+CMSParallelInitialMarkEnabled
+#-XX:+CMSEdenChunksRecordAlways
+## some JVMs will fill up their heap when accessed via JMX, see CASSANDRA-6541
+#-XX:+CMSClassUnloadingEnabled
+
+
+
+### G1 Settings
+## Use the Hotspot garbage-first collector.
+-XX:+UseG1GC
+-XX:+ParallelRefProcEnabled
+
+#
+## Have the JVM do less remembered set work during STW, instead
+## preferring concurrent GC. Reduces p99.9 latency.
+-XX:G1RSetUpdatingPauseTimePercent=5
+#
+## Main G1GC tunable: lowering the pause target will lower throughput and vise versa.
+## 200ms is the JVM default and lowest viable setting
+## 1000ms increases throughput. Keep it smaller than the timeouts in cassandra.yaml.
+-XX:MaxGCPauseMillis=500
+
+## Optional G1 Settings
+# Save CPU time on large (>= 16GB) heaps by delaying region scanning
+# until the heap is 70% full. The default in Hotspot 8u40 is 40%.
+#-XX:InitiatingHeapOccupancyPercent=70
+
+# For systems with > 8 cores, the default ParallelGCThreads is 5/8 the number of logical cores.
+# Otherwise equal to the number of cores when 8 or less.
+# Machines with > 10 cores should try setting these to <= full cores.
+#-XX:ParallelGCThreads=16
+# By default, ConcGCThreads is 1/4 of ParallelGCThreads.
+# Setting both to the same value can reduce STW durations.
+#-XX:ConcGCThreads=16
+
+
+### JPMS
+
+-Djdk.attach.allowAttachSelf=true
+--add-exports java.base/jdk.internal.misc=ALL-UNNAMED
+--add-exports java.base/jdk.internal.ref=ALL-UNNAMED
+--add-exports java.base/jdk.internal.perf=ALL-UNNAMED
+--add-exports java.base/sun.nio.ch=ALL-UNNAMED
+--add-exports java.management.rmi/com.sun.jmx.remote.internal.rmi=ALL-UNNAMED
+--add-exports java.rmi/sun.rmi.registry=ALL-UNNAMED
+--add-exports java.rmi/sun.rmi.server=ALL-UNNAMED
+# --add-exports java.sql/java.sql=ALL-UNNAMED --> not seen in
+# Below exist in the other script
+--add-exports jdk.naming.dns/com.sun.jndi.dns=java.naming
+--add-exports jdk.unsupported/sun.misc=ALL-UNNAMED
+--add-exports jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED
+
+--add-opens java.base/jdk.internal.loader=ALL-UNNAMED
+--add-opens java.base/jdk.internal.ref=ALL-UNNAMED
+--add-opens java.base/jdk.internal.reflect=ALL-UNNAMED
+--add-opens java.base/jdk.internal.math=ALL-UNNAMED
+--add-opens java.base/jdk.internal.module=ALL-UNNAMED
+--add-opens jdk.management/com.sun.management.internal=ALL-UNNAMED
+# Added in
+--add-opens java.base/java.lang.module=ALL-UNNAMED
+--add-opens jdk.management.jfr/jdk.management.jfr=ALL-UNNAMED
+--add-opens jdk.naming.dns/com.sun.jndi.dns=ALL-UNNAMED
+--add-opens jdk.compiler/com.sun.tools.javac=ALL-UNNAMED
+--add-opens java.base/java.nio.file.spi=ALL-UNNAMED
+--add-opens java.base/java.net=ALL-UNNAMED
+--add-opens java.base/jdk.internal.math=ALL-UNNAMED
+--add-opens java.base/java.lang.reflect=ALL-UNNAMED
+
+# required for org.apache.cassandra.Util.getSupportedMTimeGranularity
+--add-opens java.base/java.nio.file.attribute=ALL-UNNAMED
+
+
+### GC logging options -- uncomment to enable
+
+# Java 11 (and newer) GC logging options:
+# See description of https://bugs.openjdk.java.net/browse/JDK-8046148 for details about the syntax
+# The following is the equivalent to -XX:+PrintGCDetails -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10 -XX:GCLogFileSize=10M
+# -Xlog:gc=info,heap*=debug,age*=debug,safepoint=info,promotion*=debug:file=/var/log/cassandra/gc.log:time,uptime,pid,tid,level:filecount=10,filesize=10485760
+
+# Notes for Java 8 migration:
+#
+# -XX:+PrintGCDetails                   maps to -Xlog:gc*:... - i.e. add a '*' after "gc"
+# -XX:+PrintGCDateStamps                maps to decorator 'time'
+#
+# -XX:+PrintHeapAtGC                    maps to 'heap' with level 'trace'
+# -XX:+PrintTenuringDistribution        maps to 'age' with level 'debug'
+# -XX:+PrintGCApplicationStoppedTime    maps to 'safepoint' with level 'info'
+# -XX:+PrintPromotionFailure            maps to 'promotion' with level 'trace'
+# -XX:PrintFLSStatistics=1              maps to 'freelist' with level 'trace'
+
+### Netty Options
+
+# On Java >= 9 Netty requires the io.netty.tryReflectionSetAccessible system property to be set to true to enable
+# creation of direct buffers using Unsafe. Without it, this falls back to ByteBuffer.allocateDirect which has
+# inferior performance and risks exceeding MaxDirectMemory
+-Dio.netty.tryReflectionSetAccessible=true
+
+### Enable vector incubator feature (simd support)
+
+--add-modules jdk.incubator.vector
+
+### Compatibility Options
+--add-opens java.base/java.io=ALL-UNNAMED --add-opens java.base/sun.nio.ch=ALL-UNNAMED --add-opens java.base/java.nio=ALL-UNNAMED --add-opens java.base/java.lang=ALL-UNNAMED --add-opens java.base/java.util=ALL-UNNAMED
+-Djava.security.manager=allow
+
+# The newline in the end of file is intentional
\ No newline at end of file
diff --git a/ide/idea/workspace.xml b/ide/idea/workspace.xml
index 8337c5eee7b9..6e38eef56033 100644
--- a/ide/idea/workspace.xml
+++ b/ide/idea/workspace.xml
@@ -151,6 +151,7 @@
                                           -Dcassandra.storagedir=$PROJECT_DIR$/data
                                           -Djava.library.path=$PROJECT_DIR$/lib/sigar-bin
                                           -Dlogback.configurationFile=file://$PROJECT_DIR$/conf/logback.xml
+                                          -Dnet.bytebuddy.experimental=true
                                           -XX:HeapDumpPath=build/test
                                           -ea" />
       <option name="PROGRAM_PARAMETERS" value="" />
@@ -203,6 +204,7 @@
                                           -XX:HeapDumpPath=build/test
                                           -XX:MaxMetaspaceSize=2G
                                           -XX:SoftRefLRUPolicyMSPerMB=0
+                                          -Dnet.bytebuddy.experimental=true
                                           -Xmx4G
                                           -ea" />
       <option name="PARAMETERS" value="" />
@@ -234,6 +236,7 @@
                                           -Dcassandra.triggers_dir=$PROJECT_DIR$/conf/triggers
                                           -Djava.library.path=$PROJECT_DIR$/lib/sigar-bin
                                           -Dlogback.configurationFile=file://$PROJECT_DIR$/conf/logback.xml
+                                          -Dnet.bytebuddy.experimental=true
                                           -XX:HeapDumpPath=build/test
                                           -Xmx1G
                                           -ea" />
diff --git a/src/java/org/apache/cassandra/db/memtable/TrieMemtable.java b/src/java/org/apache/cassandra/db/memtable/TrieMemtable.java
index d08fc1635773..4b98ffaa8802 100644
--- a/src/java/org/apache/cassandra/db/memtable/TrieMemtable.java
+++ b/src/java/org/apache/cassandra/db/memtable/TrieMemtable.java
@@ -90,7 +90,7 @@ public class TrieMemtable extends AbstractShardedMemtable
     private static final Logger logger = LoggerFactory.getLogger(TrieMemtable.class);
 
     /** Buffer type to use for memtable tries (on- vs off-heap) */
-    public static final BufferType BUFFER_TYPE = DatabaseDescriptor.getMemtableAllocationType().toBufferType();
+    public static BufferType BUFFER_TYPE = DatabaseDescriptor.getMemtableAllocationType().toBufferType();
 
     /** If keys is below this length, we will use a recursive procedure for inserting data in the memtable trie. */
     @VisibleForTesting
diff --git a/src/java/org/apache/cassandra/index/sai/disk/io/IndexFileUtils.java b/src/java/org/apache/cassandra/index/sai/disk/io/IndexFileUtils.java
index 6ef529416236..e50527dc5669 100644
--- a/src/java/org/apache/cassandra/index/sai/disk/io/IndexFileUtils.java
+++ b/src/java/org/apache/cassandra/index/sai/disk/io/IndexFileUtils.java
@@ -49,8 +49,9 @@ public class IndexFileUtils
 {
     protected static final Logger logger = LoggerFactory.getLogger(IndexFileUtils.class);
 
+    // KATE non-final only because we need to set it for testing...
     @VisibleForTesting
-    public static final SequentialWriterOption DEFAULT_WRITER_OPTION = SequentialWriterOption.newBuilder()
+    public static SequentialWriterOption DEFAULT_WRITER_OPTION = SequentialWriterOption.newBuilder()
                                                                                              .trickleFsync(DatabaseDescriptor.getTrickleFsync())
                                                                                              .trickleFsyncByteInterval(DatabaseDescriptor.getTrickleFsyncIntervalInKiB() * 1024)
                                                                                              .bufferType(BufferType.OFF_HEAP)
@@ -62,7 +63,8 @@ public class IndexFileUtils
     private static final Supplier<Checksum> LEGACY_CHECKSUM_FACTORY = CRC32::new;
     private static IndexFileUtils overrideInstance = null;
 
-    private final SequentialWriterOption writerOption;
+    // KATE non-final only because we need to set it for testing...
+    private SequentialWriterOption writerOption;
 
     public static synchronized void setOverrideInstance(IndexFileUtils overrideInstance)
     {
diff --git a/src/java/org/apache/cassandra/index/sai/disk/v1/V1OnDiskFormat.java b/src/java/org/apache/cassandra/index/sai/disk/v1/V1OnDiskFormat.java
index 541f8723f047..34bc750cd13d 100644
--- a/src/java/org/apache/cassandra/index/sai/disk/v1/V1OnDiskFormat.java
+++ b/src/java/org/apache/cassandra/index/sai/disk/v1/V1OnDiskFormat.java
@@ -100,7 +100,7 @@ public class V1OnDiskFormat implements OnDiskFormat
      * ex. If there is one column index building per table across 8 compactors, each index will be
      *     eligible to flush once it reaches (segment_write_buffer_space_mb / 8) MBs.
      */
-    public static final long SEGMENT_BUILD_MEMORY_LIMIT = 1024L * 1024L * DatabaseDescriptor.getSAISegmentWriteBufferSpace();
+    public static long SEGMENT_BUILD_MEMORY_LIMIT = 1024L * 1024L * DatabaseDescriptor.getSAISegmentWriteBufferSpace();
 
     public static final NamedMemoryLimiter SEGMENT_BUILD_MEMORY_LIMITER =
     new NamedMemoryLimiter(SEGMENT_BUILD_MEMORY_LIMIT, "SSTable-attached Index Segment Builder");
diff --git a/src/java/org/apache/cassandra/index/sai/plan/QueryController.java b/src/java/org/apache/cassandra/index/sai/plan/QueryController.java
index 839645ac2ae3..b2ae2b7cea85 100644
--- a/src/java/org/apache/cassandra/index/sai/plan/QueryController.java
+++ b/src/java/org/apache/cassandra/index/sai/plan/QueryController.java
@@ -106,7 +106,7 @@
 public class QueryController implements Plan.Executor, Plan.CostEstimator
 {
     private static final Logger logger = LoggerFactory.getLogger(QueryController.class);
-
+    
     /**
      * Controls whether we optimize query plans.
      * 0 disables the optimizer. As a side effect, hybrid ANN queries will default to FilterSortOrder.SCAN_THEN_FILTER.
diff --git a/src/java/org/apache/cassandra/index/sai/utils/RangeIntersectionIterator.java b/src/java/org/apache/cassandra/index/sai/utils/RangeIntersectionIterator.java
index ae7bf454681b..e1dd8bfe6125 100644
--- a/src/java/org/apache/cassandra/index/sai/utils/RangeIntersectionIterator.java
+++ b/src/java/org/apache/cassandra/index/sai/utils/RangeIntersectionIterator.java
@@ -42,7 +42,7 @@ public class RangeIntersectionIterator extends RangeIterator
 
     // The cassandra.sai.intersection_clause_limit (default: 2) controls the maximum number of range iterator that
     // will be used in the final intersection of a query operation.
-    public static final int INTERSECTION_CLAUSE_LIMIT = SAI_INTERSECTION_CLAUSE_LIMIT.getInt();
+    public static int INTERSECTION_CLAUSE_LIMIT = SAI_INTERSECTION_CLAUSE_LIMIT.getInt();
 
     static
     {
diff --git a/src/java/org/apache/cassandra/index/sasi/utils/trie/AbstractPatriciaTrie.java b/src/java/org/apache/cassandra/index/sasi/utils/trie/AbstractPatriciaTrie.java
index 8067ccc30cdc..8b89eaee4a45 100644
--- a/src/java/org/apache/cassandra/index/sasi/utils/trie/AbstractPatriciaTrie.java
+++ b/src/java/org/apache/cassandra/index/sasi/utils/trie/AbstractPatriciaTrie.java
@@ -790,7 +790,7 @@ TrieEntry<K, V> nextEntryImpl(TrieEntry<K, V> start, TrieEntry<K, V> previous, T
      * This is implemented by going always to the left until
      * we encounter a valid uplink. That uplink is the first key.
      */
-    TrieEntry<K, V> firstEntry()
+    public TrieEntry<K, V> firstEntry()
     {
         // if Trie is empty, no first node.
         return isEmpty() ? null : followLeft(root);
diff --git a/src/java/org/apache/cassandra/index/sasi/utils/trie/PatriciaTrie.java b/src/java/org/apache/cassandra/index/sasi/utils/trie/PatriciaTrie.java
index a36af9828c8b..9bf3b1754e8a 100644
--- a/src/java/org/apache/cassandra/index/sasi/utils/trie/PatriciaTrie.java
+++ b/src/java/org/apache/cassandra/index/sasi/utils/trie/PatriciaTrie.java
@@ -414,7 +414,7 @@ private TrieEntry<K, V> subtree(K prefix)
      * <p>This is implemented by going always to the right until
      * we encounter a valid uplink. That uplink is the last key.
      */
-    private TrieEntry<K, V> lastEntry()
+    public TrieEntry<K, V> lastEntry()
     {
         return followRight(root.left);
     }
diff --git a/src/java/org/apache/cassandra/utils/btree/BTreeSet.java b/src/java/org/apache/cassandra/utils/btree/BTreeSet.java
index b20729d7da8a..2824e348fc3d 100644
--- a/src/java/org/apache/cassandra/utils/btree/BTreeSet.java
+++ b/src/java/org/apache/cassandra/utils/btree/BTreeSet.java
@@ -312,6 +312,41 @@ public ListIterator<V> listIterator(int index)
         throw new UnsupportedOperationException();
     }
 
+    public BTreeSet<V> reversed()
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    public V removeLast()
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    public V removeFirst()
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    public V getLast()
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    public V getFirst()
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    public void addLast(V v)
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    public void addFirst(V v)
+    {
+        throw new UnsupportedOperationException();
+    }
+
     public static class BTreeRange<V> extends BTreeSet<V>
     {
         // both inclusive
diff --git a/test/unit/org/apache/cassandra/index/sai/SAITester.java b/test/unit/org/apache/cassandra/index/sai/SAITester.java
index d1d46d1f7c18..ce82c9fafd66 100644
--- a/test/unit/org/apache/cassandra/index/sai/SAITester.java
+++ b/test/unit/org/apache/cassandra/index/sai/SAITester.java
@@ -92,6 +92,7 @@
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.service.snapshot.TableSnapshot;
 import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.ReflectionUtils;
 import org.apache.cassandra.utils.Throwables;
 import org.apache.lucene.codecs.CodecUtil;
 import org.awaitility.Awaitility;
@@ -898,7 +899,7 @@ protected static void setSegmentWriteBufferSpace(final int segmentSize) throws E
         NamedMemoryLimiter limiter = (NamedMemoryLimiter) V1OnDiskFormat.class.getDeclaredField("SEGMENT_BUILD_MEMORY_LIMITER").get(null);
         Field limitBytes = limiter.getClass().getDeclaredField("limitBytes");
         limitBytes.setAccessible(true);
-        Field modifiersField = Field.class.getDeclaredField("modifiers");
+        Field modifiersField = ReflectionUtils.getField(Field.class, "modifiers");
         modifiersField.setAccessible(true);
         modifiersField.setInt(limitBytes, limitBytes.getModifiers() & ~Modifier.FINAL);
         limitBytes.set(limiter, segmentSize);
diff --git a/test/unit/org/apache/cassandra/index/sai/SAIUtil.java b/test/unit/org/apache/cassandra/index/sai/SAIUtil.java
index 8f56a0f4ca9b..f7580332077c 100644
--- a/test/unit/org/apache/cassandra/index/sai/SAIUtil.java
+++ b/test/unit/org/apache/cassandra/index/sai/SAIUtil.java
@@ -21,6 +21,7 @@
 import java.lang.reflect.Field;
 
 import org.apache.cassandra.index.sai.disk.format.Version;
+import org.apache.cassandra.utils.ReflectionUtils;
 
 public class SAIUtil
 {
@@ -31,7 +32,7 @@ public static void setLatestVersion(Version version)
         {
             latest = Version.class.getDeclaredField("LATEST");
             latest.setAccessible(true);
-            Field modifiersField = Field.class.getDeclaredField("modifiers");
+            Field modifiersField = ReflectionUtils.getField(Field.class, "modifiers");
             modifiersField.setAccessible(true);
             latest.set(null, version);
         }
diff --git a/test/unit/org/apache/cassandra/index/sai/cql/VectorUpdateDeleteTest.java b/test/unit/org/apache/cassandra/index/sai/cql/VectorUpdateDeleteTest.java
index 5b74719c87fb..97f4bc5ef2f3 100644
--- a/test/unit/org/apache/cassandra/index/sai/cql/VectorUpdateDeleteTest.java
+++ b/test/unit/org/apache/cassandra/index/sai/cql/VectorUpdateDeleteTest.java
@@ -34,6 +34,7 @@
 import org.apache.cassandra.index.sai.disk.format.Version;
 import org.apache.cassandra.index.sai.disk.vector.VectorMemtableIndex;
 import org.apache.cassandra.index.sai.plan.QueryController;
+import org.apache.cassandra.utils.ReflectionUtils;
 
 import static org.apache.cassandra.index.sai.cql.VectorTypeTest.assertContainsInt;
 import static org.apache.cassandra.index.sai.disk.vector.CassandraOnHeapGraph.MIN_PQ_ROWS;
diff --git a/test/unit/org/apache/cassandra/index/sai/disk/io/TrackingIndexFileUtils.java b/test/unit/org/apache/cassandra/index/sai/disk/io/TrackingIndexFileUtils.java
index be9bca5c3300..53ed1295e899 100644
--- a/test/unit/org/apache/cassandra/index/sai/disk/io/TrackingIndexFileUtils.java
+++ b/test/unit/org/apache/cassandra/index/sai/disk/io/TrackingIndexFileUtils.java
@@ -29,6 +29,7 @@
 
 import org.apache.cassandra.io.util.FileHandle;
 import org.apache.cassandra.io.util.SequentialWriterOption;
+import org.apache.cassandra.utils.ReflectionUtils;
 
 public class TrackingIndexFileUtils extends IndexFileUtils
 {
diff --git a/test/unit/org/apache/cassandra/index/sai/memory/TrieMemtableIndexTest.java b/test/unit/org/apache/cassandra/index/sai/memory/TrieMemtableIndexTest.java
index b6995a64ba88..8184bea22c8c 100644
--- a/test/unit/org/apache/cassandra/index/sai/memory/TrieMemtableIndexTest.java
+++ b/test/unit/org/apache/cassandra/index/sai/memory/TrieMemtableIndexTest.java
@@ -67,6 +67,7 @@
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.Pair;
+import org.apache.cassandra.utils.ReflectionUtils;
 import org.apache.cassandra.utils.bytecomparable.ByteComparable;
 import org.apache.cassandra.utils.bytecomparable.ByteSource;
 import org.apache.cassandra.utils.concurrent.OpOrder;
@@ -364,7 +365,7 @@ private void setTrieMemtableBufferType(final BufferType newBufferType) throws Ex
     {
         Field bufferType = TrieMemtable.class.getDeclaredField("BUFFER_TYPE");
         bufferType.setAccessible(true);
-        Field modifiersField = Field.class.getDeclaredField("modifiers");
+        Field modifiersField = ReflectionUtils.getField(Field.class, "modifiers");
         modifiersField.setAccessible(true);
         modifiersField.setInt(bufferType, bufferType.getModifiers() & ~Modifier.FINAL);
         bufferType.set(null, newBufferType);
diff --git a/test/unit/org/apache/cassandra/transport/MessagePayloadTest.java b/test/unit/org/apache/cassandra/transport/MessagePayloadTest.java
index aebd13522d71..c922b8fbb292 100644
--- a/test/unit/org/apache/cassandra/transport/MessagePayloadTest.java
+++ b/test/unit/org/apache/cassandra/transport/MessagePayloadTest.java
@@ -17,14 +17,11 @@
  */
 package org.apache.cassandra.transport;
 
-import java.lang.reflect.Field;
-import java.lang.reflect.Modifier;
 import java.nio.ByteBuffer;
 import java.util.Collections;
 import java.util.Map;
 
 import org.junit.After;
-import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Test;
@@ -47,7 +44,6 @@
 import org.apache.cassandra.transport.messages.QueryMessage;
 import org.apache.cassandra.transport.messages.ResultMessage;
 import org.apache.cassandra.utils.MD5Digest;
-import org.apache.cassandra.utils.ReflectionUtils;
 
 import static org.apache.cassandra.utils.ByteBufferUtil.bytes;
 
@@ -56,47 +52,11 @@ public class MessagePayloadTest extends CQLTester
     public static Map<String, ByteBuffer> requestPayload;
     public static Map<String, ByteBuffer> responsePayload;
 
-    private static Field cqlQueryHandlerField;
-    private static boolean modifiersAccessible;
-
     @BeforeClass
-    public static void makeCqlQueryHandlerAccessible()
+    public static void setUpClass()
     {
-        try
-        {
-            cqlQueryHandlerField = ClientState.class.getDeclaredField("cqlQueryHandler");
-            cqlQueryHandlerField.setAccessible(true);
-
-            Field modifiersField = ReflectionUtils.getModifiersField();
-            modifiersAccessible = modifiersField.isAccessible();
-            modifiersField.setAccessible(true);
-            modifiersField.setInt(cqlQueryHandlerField, cqlQueryHandlerField.getModifiers() & ~Modifier.FINAL);
-        }
-        catch (IllegalAccessException | NoSuchFieldException e)
-        {
-            throw new RuntimeException(e);
-        }
-    }
-
-    @AfterClass
-    public static void resetCqlQueryHandlerField()
-    {
-        if (cqlQueryHandlerField == null)
-            return;
-        try
-        {
-            Field modifiersField = ReflectionUtils.getModifiersField();
-            modifiersField.setAccessible(true);
-            modifiersField.setInt(cqlQueryHandlerField, cqlQueryHandlerField.getModifiers() | Modifier.FINAL);
-
-            cqlQueryHandlerField.setAccessible(false);
-
-            modifiersField.setAccessible(modifiersAccessible);
-        }
-        catch (IllegalAccessException | NoSuchFieldException e)
-        {
-            throw new RuntimeException(e);
-        }
+        System.setProperty("cassandra.custom_query_handler_class", "org.apache.cassandra.transport.MessagePayloadTest$TestQueryHandler");
+        CQLTester.setUpClass();
     }
 
     @After
@@ -115,256 +75,226 @@ public void dropCreatedTable()
     @Test
     public void testMessagePayloadBeta() throws Throwable
     {
-        QueryHandler queryHandler = (QueryHandler) cqlQueryHandlerField.get(null);
-        cqlQueryHandlerField.set(null, new TestQueryHandler());
+        requireNetwork();
+
+        SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(),
+                nativePort,
+                ProtocolVersion.V5,
+                true,
+                new EncryptionOptions());
         try
         {
-            requireNetwork();
-
-            Assert.assertSame(TestQueryHandler.class, ClientState.getCQLQueryHandler().getClass());
+            client.connect(false);
+
+            Map<String, ByteBuffer> reqMap;
+            Map<String, ByteBuffer> respMap;
+
+            QueryOptions queryOptions = QueryOptions.create(
+                    QueryOptions.DEFAULT.getConsistency(),
+                    QueryOptions.DEFAULT.getValues(),
+                    QueryOptions.DEFAULT.skipMetadata(),
+                    QueryOptions.DEFAULT.getPageSize(),
+                    QueryOptions.DEFAULT.getPagingState(),
+                    QueryOptions.DEFAULT.getSerialConsistency(null),
+                    ProtocolVersion.V5,
+                    KEYSPACE);
+            QueryMessage queryMessage = new QueryMessage("CREATE TABLE atable (pk int PRIMARY KEY, v text)",
+                    queryOptions);
+            PrepareMessage prepareMessage = new PrepareMessage("SELECT * FROM atable", KEYSPACE);
+
+            reqMap = Collections.singletonMap("foo", bytes(42));
+            responsePayload = respMap = Collections.singletonMap("bar", bytes(42));
+            queryMessage.setCustomPayload(reqMap);
+            Message.Response queryResponse = client.execute(queryMessage);
+            payloadEquals(reqMap, requestPayload);
+            payloadEquals(respMap, queryResponse.getCustomPayload());
+
+            reqMap = Collections.singletonMap("foo", bytes(43));
+            responsePayload = respMap = Collections.singletonMap("bar", bytes(43));
+            prepareMessage.setCustomPayload(reqMap);
+            ResultMessage.Prepared prepareResponse = (ResultMessage.Prepared) client.execute(prepareMessage);
+            payloadEquals(reqMap, requestPayload);
+            payloadEquals(respMap, prepareResponse.getCustomPayload());
+
+            ExecuteMessage executeMessage = new ExecuteMessage(prepareResponse.statementId, prepareResponse.resultMetadataId, QueryOptions.DEFAULT);
+            reqMap = Collections.singletonMap("foo", bytes(44));
+            responsePayload = respMap = Collections.singletonMap("bar", bytes(44));
+            executeMessage.setCustomPayload(reqMap);
+            Message.Response executeResponse = client.execute(executeMessage);
+            payloadEquals(reqMap, requestPayload);
+            payloadEquals(respMap, executeResponse.getCustomPayload());
+
+            BatchMessage batchMessage = new BatchMessage(BatchStatement.Type.UNLOGGED,
+                    Collections.<Object>singletonList("INSERT INTO " + KEYSPACE + ".atable (pk,v) VALUES (1, 'foo')"),
+                    Collections.singletonList(Collections.<ByteBuffer>emptyList()),
+                    queryOptions);
+            reqMap = Collections.singletonMap("foo", bytes(45));
+            responsePayload = respMap = Collections.singletonMap("bar", bytes(45));
+            batchMessage.setCustomPayload(reqMap);
+            Message.Response batchResponse = client.execute(batchMessage);
+            payloadEquals(reqMap, requestPayload);
+            payloadEquals(respMap, batchResponse.getCustomPayload());
 
-            SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(),
-                                                   nativePort,
-                                                   ProtocolVersion.V5,
-                                                   true,
-                                                   new EncryptionOptions());
-            try
-            {
-                client.connect(false);
-
-                Map<String, ByteBuffer> reqMap;
-                Map<String, ByteBuffer> respMap;
-
-                QueryOptions queryOptions = QueryOptions.create(
-                  QueryOptions.DEFAULT.getConsistency(),
-                  QueryOptions.DEFAULT.getValues(),
-                  QueryOptions.DEFAULT.skipMetadata(),
-                  QueryOptions.DEFAULT.getPageSize(),
-                  QueryOptions.DEFAULT.getPagingState(),
-                  QueryOptions.DEFAULT.getSerialConsistency(null),
-                  ProtocolVersion.V5,
-                  KEYSPACE);
-                QueryMessage queryMessage = new QueryMessage("CREATE TABLE atable (pk int PRIMARY KEY, v text)",
-                                                             queryOptions);
-                PrepareMessage prepareMessage = new PrepareMessage("SELECT * FROM atable", KEYSPACE);
-
-                reqMap = Collections.singletonMap("foo", bytes(42));
-                responsePayload = respMap = Collections.singletonMap("bar", bytes(42));
-                queryMessage.setCustomPayload(reqMap);
-                Message.Response queryResponse = client.execute(queryMessage);
-                payloadEquals(reqMap, requestPayload);
-                payloadEquals(respMap, queryResponse.getCustomPayload());
-
-                reqMap = Collections.singletonMap("foo", bytes(43));
-                responsePayload = respMap = Collections.singletonMap("bar", bytes(43));
-                prepareMessage.setCustomPayload(reqMap);
-                ResultMessage.Prepared prepareResponse = (ResultMessage.Prepared) client.execute(prepareMessage);
-                payloadEquals(reqMap, requestPayload);
-                payloadEquals(respMap, prepareResponse.getCustomPayload());
-
-                ExecuteMessage executeMessage = new ExecuteMessage(prepareResponse.statementId, prepareResponse.resultMetadataId, QueryOptions.DEFAULT);
-                reqMap = Collections.singletonMap("foo", bytes(44));
-                responsePayload = respMap = Collections.singletonMap("bar", bytes(44));
-                executeMessage.setCustomPayload(reqMap);
-                Message.Response executeResponse = client.execute(executeMessage);
-                payloadEquals(reqMap, requestPayload);
-                payloadEquals(respMap, executeResponse.getCustomPayload());
-
-                BatchMessage batchMessage = new BatchMessage(BatchStatement.Type.UNLOGGED,
-                                                             Collections.<Object>singletonList("INSERT INTO " + KEYSPACE + ".atable (pk,v) VALUES (1, 'foo')"),
-                                                             Collections.singletonList(Collections.<ByteBuffer>emptyList()),
-                                                             queryOptions);
-                reqMap = Collections.singletonMap("foo", bytes(45));
-                responsePayload = respMap = Collections.singletonMap("bar", bytes(45));
-                batchMessage.setCustomPayload(reqMap);
-                Message.Response batchResponse = client.execute(batchMessage);
-                payloadEquals(reqMap, requestPayload);
-                payloadEquals(respMap, batchResponse.getCustomPayload());
-            }
-            finally
-            {
-                client.close();
-            }
         }
         finally
         {
-            cqlQueryHandlerField.set(null, queryHandler);
+            client.close();
         }
     }
 
     @Test
     public void testMessagePayload() throws Throwable
     {
-        QueryHandler queryHandler = (QueryHandler) cqlQueryHandlerField.get(null);
-        cqlQueryHandlerField.set(null, new TestQueryHandler());
-        try
-        {
-            requireNetwork();
+        requireNetwork();
 
-            Assert.assertSame(TestQueryHandler.class, ClientState.getCQLQueryHandler().getClass());
+        Assert.assertSame(TestQueryHandler.class, ClientState.getCQLQueryHandler().getClass());
 
-            SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), nativePort);
-            try
-            {
-                client.connect(false);
-
-                Map<String, ByteBuffer> reqMap;
-                Map<String, ByteBuffer> respMap;
-
-                QueryMessage queryMessage = new QueryMessage(
-                                                            "CREATE TABLE " + KEYSPACE + ".atable (pk int PRIMARY KEY, v text)",
-                                                            QueryOptions.DEFAULT
-                );
-                PrepareMessage prepareMessage = new PrepareMessage("SELECT * FROM " + KEYSPACE + ".atable", null);
-
-                reqMap = Collections.singletonMap("foo", bytes(42));
-                responsePayload = respMap = Collections.singletonMap("bar", bytes(42));
-                queryMessage.setCustomPayload(reqMap);
-                Message.Response queryResponse = client.execute(queryMessage);
-                payloadEquals(reqMap, requestPayload);
-                payloadEquals(respMap, queryResponse.getCustomPayload());
-
-                reqMap = Collections.singletonMap("foo", bytes(43));
-                responsePayload = respMap = Collections.singletonMap("bar", bytes(43));
-                prepareMessage.setCustomPayload(reqMap);
-                ResultMessage.Prepared prepareResponse = (ResultMessage.Prepared) client.execute(prepareMessage);
-                payloadEquals(reqMap, requestPayload);
-                payloadEquals(respMap, prepareResponse.getCustomPayload());
-
-                ExecuteMessage executeMessage = new ExecuteMessage(prepareResponse.statementId, prepareResponse.resultMetadataId, QueryOptions.DEFAULT);
-                reqMap = Collections.singletonMap("foo", bytes(44));
-                responsePayload = respMap = Collections.singletonMap("bar", bytes(44));
-                executeMessage.setCustomPayload(reqMap);
-                Message.Response executeResponse = client.execute(executeMessage);
-                payloadEquals(reqMap, requestPayload);
-                payloadEquals(respMap, executeResponse.getCustomPayload());
-
-                BatchMessage batchMessage = new BatchMessage(BatchStatement.Type.UNLOGGED,
-                                                             Collections.<Object>singletonList("INSERT INTO " + KEYSPACE + ".atable (pk,v) VALUES (1, 'foo')"),
-                                                             Collections.singletonList(Collections.<ByteBuffer>emptyList()),
-                                                             QueryOptions.DEFAULT);
-                reqMap = Collections.singletonMap("foo", bytes(45));
-                responsePayload = respMap = Collections.singletonMap("bar", bytes(45));
-                batchMessage.setCustomPayload(reqMap);
-                Message.Response batchResponse = client.execute(batchMessage);
-                payloadEquals(reqMap, requestPayload);
-                payloadEquals(respMap, batchResponse.getCustomPayload());
-            }
-            finally
-            {
-                client.close();
-            }
+        SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), nativePort);
+        try
+        {
+            client.connect(false);
+
+            Map<String, ByteBuffer> reqMap;
+            Map<String, ByteBuffer> respMap;
+
+            QueryMessage queryMessage = new QueryMessage(
+                    "CREATE TABLE " + KEYSPACE + ".atable (pk int PRIMARY KEY, v text)",
+                    QueryOptions.DEFAULT
+            );
+            PrepareMessage prepareMessage = new PrepareMessage("SELECT * FROM " + KEYSPACE + ".atable", null);
+
+            reqMap = Collections.singletonMap("foo", bytes(42));
+            responsePayload = respMap = Collections.singletonMap("bar", bytes(42));
+            queryMessage.setCustomPayload(reqMap);
+            Message.Response queryResponse = client.execute(queryMessage);
+            payloadEquals(reqMap, requestPayload);
+            payloadEquals(respMap, queryResponse.getCustomPayload());
+
+            reqMap = Collections.singletonMap("foo", bytes(43));
+            responsePayload = respMap = Collections.singletonMap("bar", bytes(43));
+            prepareMessage.setCustomPayload(reqMap);
+            ResultMessage.Prepared prepareResponse = (ResultMessage.Prepared) client.execute(prepareMessage);
+            payloadEquals(reqMap, requestPayload);
+            payloadEquals(respMap, prepareResponse.getCustomPayload());
+
+            ExecuteMessage executeMessage = new ExecuteMessage(prepareResponse.statementId, prepareResponse.resultMetadataId, QueryOptions.DEFAULT);
+            reqMap = Collections.singletonMap("foo", bytes(44));
+            responsePayload = respMap = Collections.singletonMap("bar", bytes(44));
+            executeMessage.setCustomPayload(reqMap);
+            Message.Response executeResponse = client.execute(executeMessage);
+            payloadEquals(reqMap, requestPayload);
+            payloadEquals(respMap, executeResponse.getCustomPayload());
+
+            BatchMessage batchMessage = new BatchMessage(BatchStatement.Type.UNLOGGED,
+                    Collections.<Object>singletonList("INSERT INTO " + KEYSPACE + ".atable (pk,v) VALUES (1, 'foo')"),
+                    Collections.singletonList(Collections.<ByteBuffer>emptyList()),
+                    QueryOptions.DEFAULT);
+            reqMap = Collections.singletonMap("foo", bytes(45));
+            responsePayload = respMap = Collections.singletonMap("bar", bytes(45));
+            batchMessage.setCustomPayload(reqMap);
+            Message.Response batchResponse = client.execute(batchMessage);
+            payloadEquals(reqMap, requestPayload);
+            payloadEquals(respMap, batchResponse.getCustomPayload());
         }
         finally
         {
-            cqlQueryHandlerField.set(null, queryHandler);
+            client.close();
         }
     }
 
     @Test
     public void testMessagePayloadVersion3() throws Throwable
     {
-        QueryHandler queryHandler = (QueryHandler) cqlQueryHandlerField.get(null);
-        cqlQueryHandlerField.set(null, new TestQueryHandler());
+        requireNetwork();
+
+        Assert.assertSame(TestQueryHandler.class, ClientState.getCQLQueryHandler().getClass());
+
+        SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), nativePort, ProtocolVersion.V3);
         try
         {
-            requireNetwork();
+            client.connect(false);
 
-            Assert.assertSame(TestQueryHandler.class, ClientState.getCQLQueryHandler().getClass());
+            Map<String, ByteBuffer> reqMap;
 
-            SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), nativePort, ProtocolVersion.V3);
+            QueryMessage queryMessage = new QueryMessage(
+                    "CREATE TABLE " + KEYSPACE + ".atable (pk int PRIMARY KEY, v text)",
+                    QueryOptions.DEFAULT
+            );
+            PrepareMessage prepareMessage = new PrepareMessage("SELECT * FROM " + KEYSPACE + ".atable", null);
+
+            reqMap = Collections.singletonMap("foo", bytes(42));
+            responsePayload = Collections.singletonMap("bar", bytes(42));
+            queryMessage.setCustomPayload(reqMap);
             try
             {
-                client.connect(false);
-
-                Map<String, ByteBuffer> reqMap;
-
-                QueryMessage queryMessage = new QueryMessage(
-                                                            "CREATE TABLE " + KEYSPACE + ".atable (pk int PRIMARY KEY, v text)",
-                                                            QueryOptions.DEFAULT
-                );
-                PrepareMessage prepareMessage = new PrepareMessage("SELECT * FROM " + KEYSPACE + ".atable", null);
-
-                reqMap = Collections.singletonMap("foo", bytes(42));
-                responsePayload = Collections.singletonMap("bar", bytes(42));
-                queryMessage.setCustomPayload(reqMap);
-                try
-                {
-                    client.execute(queryMessage);
-                    Assert.fail();
-                }
-                catch (RuntimeException e)
-                {
-                    Assert.assertTrue(e.getCause() instanceof ProtocolException);
-                }
-                // when a protocol exception is thrown by the server the connection is closed, so need to re-connect
-                client.close();
-                client.connect(false);
-
-                queryMessage.setCustomPayload(null);
                 client.execute(queryMessage);
+                Assert.fail();
+            }
+            catch (RuntimeException e)
+            {
+                Assert.assertTrue(e.getCause() instanceof ProtocolException);
+            }
+            // when a protocol exception is thrown by the server the connection is closed, so need to re-connect
+            client.close();
+            client.connect(false);
+
+            queryMessage.setCustomPayload(null);
+            client.execute(queryMessage);
+            reqMap = Collections.singletonMap("foo", bytes(43));
+            responsePayload = Collections.singletonMap("bar", bytes(43));
+            prepareMessage.setCustomPayload(reqMap);
+            try
+            {
+                client.execute(prepareMessage);
+                Assert.fail();
+            }
+            catch (RuntimeException e)
+            {
+                Assert.assertTrue(e.getCause() instanceof ProtocolException);
+            }
+            // when a protocol exception is thrown by the server the connection is closed, so need to re-connect
+            client.close();
+            client.connect(false);
 
-                reqMap = Collections.singletonMap("foo", bytes(43));
-                responsePayload = Collections.singletonMap("bar", bytes(43));
-                prepareMessage.setCustomPayload(reqMap);
-                try
-                {
-                    client.execute(prepareMessage);
-                    Assert.fail();
-                }
-                catch (RuntimeException e)
-                {
-                    Assert.assertTrue(e.getCause() instanceof ProtocolException);
-                }
-                // when a protocol exception is thrown by the server the connection is closed, so need to re-connect
-                client.close();
-                client.connect(false);
-
-                prepareMessage.setCustomPayload(null);
-                ResultMessage.Prepared prepareResponse = (ResultMessage.Prepared) client.execute(prepareMessage);
-
-                ExecuteMessage executeMessage = new ExecuteMessage(prepareResponse.statementId, prepareResponse.resultMetadataId, QueryOptions.DEFAULT);
-                reqMap = Collections.singletonMap("foo", bytes(44));
-                responsePayload = Collections.singletonMap("bar", bytes(44));
-                executeMessage.setCustomPayload(reqMap);
-                try
-                {
-                    client.execute(executeMessage);
-                    Assert.fail();
-                }
-                catch (RuntimeException e)
-                {
-                    Assert.assertTrue(e.getCause() instanceof ProtocolException);
-                }
-                // when a protocol exception is thrown by the server the connection is closed, so need to re-connect
-                client.close();
-                client.connect(false);
-
-                BatchMessage batchMessage = new BatchMessage(BatchStatement.Type.UNLOGGED,
-                                                             Collections.<Object>singletonList("INSERT INTO " + KEYSPACE + ".atable (pk,v) VALUES (1, 'foo')"),
-                                                             Collections.singletonList(Collections.<ByteBuffer>emptyList()),
-                                                             QueryOptions.DEFAULT);
-                reqMap = Collections.singletonMap("foo", bytes(45));
-                responsePayload = Collections.singletonMap("bar", bytes(45));
-                batchMessage.setCustomPayload(reqMap);
-                try
-                {
-                    client.execute(batchMessage);
-                    Assert.fail();
-                }
-                catch (RuntimeException e)
-                {
-                    Assert.assertTrue(e.getCause() instanceof ProtocolException);
-                }
+            prepareMessage.setCustomPayload(null);
+            ResultMessage.Prepared prepareResponse = (ResultMessage.Prepared) client.execute(prepareMessage);
+
+            ExecuteMessage executeMessage = new ExecuteMessage(prepareResponse.statementId, prepareResponse.resultMetadataId, QueryOptions.DEFAULT);
+            reqMap = Collections.singletonMap("foo", bytes(44));
+            responsePayload = Collections.singletonMap("bar", bytes(44));
+            executeMessage.setCustomPayload(reqMap);
+            try
+            {
+                client.execute(executeMessage);
+                Assert.fail();
+            }
+            catch (RuntimeException e)
+            {
+                Assert.assertTrue(e.getCause() instanceof ProtocolException);
+            }
+            // when a protocol exception is thrown by the server the connection is closed, so need to re-connect
+            client.close();
+            client.connect(false);
+
+            BatchMessage batchMessage = new BatchMessage(BatchStatement.Type.UNLOGGED,
+                    Collections.<Object>singletonList("INSERT INTO " + KEYSPACE + ".atable (pk,v) VALUES (1, 'foo')"),
+                    Collections.singletonList(Collections.<ByteBuffer>emptyList()),
+                    QueryOptions.DEFAULT);
+            reqMap = Collections.singletonMap("foo", bytes(45));
+            responsePayload = Collections.singletonMap("bar", bytes(45));
+            batchMessage.setCustomPayload(reqMap);
+            try {
+                client.execute(batchMessage);
+                Assert.fail();
             }
-            finally
+            catch(RuntimeException e)
             {
-                client.close();
+                Assert.assertTrue(e.getCause() instanceof ProtocolException);
             }
         }
         finally
         {
-            cqlQueryHandlerField.set(null, queryHandler);
+            client.close();
         }
     }
 
diff --git a/tools/bin/cassandra.in.sh b/tools/bin/cassandra.in.sh
index 555ca6ecafbc..03f8c29a860f 100644
--- a/tools/bin/cassandra.in.sh
+++ b/tools/bin/cassandra.in.sh
@@ -76,17 +76,21 @@ if [ -z $JAVA ] ; then
     exit 1;
 fi
 
+java_versions_supported=11,17,22
+
 # Determine the sort of JVM we'll be running on.
 java_ver_output=`"${JAVA:-java}" -version 2>&1`
 jvmver=`echo "$java_ver_output" | grep '[openjdk|java] version' | awk -F'"' 'NR==1 {print $2}' | cut -d\- -f1`
 JVM_VERSION=${jvmver%_*}
 short=$(echo "${jvmver}" | cut -c1-2)
 
-JAVA_VERSION=17
+JAVA_VERSION=22
 if [ "$short" = "11" ] ; then
      JAVA_VERSION=11
-elif [ "$JVM_VERSION" \< "17" ] ; then
-    echo "Cassandra 5.0 requires Java 11 or Java 17(or newer)."
+elif [ "$short" = "17" ] ; then
+      JAVA_VERSION=17
+elif [ "$JVM_VERSION" \< "22" ] ; then
+    echo "Cassandra 5.0 requires Java 11, Java 17 or Java 22(or newer)."
 fi
 
 jvm=`echo "$java_ver_output" | grep -A 1 '[openjdk|java] version' | awk 'NR==2 {print $1}'`
@@ -110,7 +114,9 @@ esac
 
 # Read user-defined JVM options from jvm-server.options file
 JVM_OPTS_FILE=$CASSANDRA_CONF/jvm${jvmoptions_variant:--clients}.options
-if [ $JAVA_VERSION -ge 17 ] ; then
+if [ $JAVA_VERSION -ge 22 ] ; then
+    JVM_DEP_OPTS_FILE=$CASSANDRA_CONF/jvm22${jvmoptions_variant:--clients}.options
+elif [ $JAVA_VERSION -ge 17 ] ; then
     JVM_DEP_OPTS_FILE=$CASSANDRA_CONF/jvm17${jvmoptions_variant:--clients}.options
 elif [ $JAVA_VERSION -ge 11 ] ; then
     JVM_DEP_OPTS_FILE=$CASSANDRA_CONF/jvm11${jvmoptions_variant:--clients}.options