CASSSIDECAR-245: Implementation of CassandraClusterSchema (#251)

Patch by Bernardo Botella; Reviewed by Francisco Guerrero for CASSSIDECAR-245

Author: bbotella

CHANGES.txt

@@ -1,5 +1,6 @@
 0.3.0
 -----
+ * Implementation of CassandraClusterSchemaMonitor (CASSSIDECAR-245)
  * Sidecar endpoint for vending statistics related to compaction (CASSSIDECAR-329)
  * Update logging dependencies (CASSSIDECAR-337)
 

conf/sidecar.yaml

@@ -107,6 +107,7 @@ sidecar:
       maximum_size: 10000
   cdc:
     segment_hardlink_cache_expiry: 5m # 5 minutes
+    table_schema_refresh_time: 60s    # Interval time between table schema refresh for CDC
   worker_pools:
     service:
       name: "sidecar-worker-pool"

gradle.properties

@@ -40,3 +40,6 @@ smallryeOpenApiVersion=4.1.1
 microprofileOpenApiVersion=3.1.1
 jakartaWsRsVersion=3.1.0
 swaggerVersion=2.2.21
+# Cdc dependencies
+kryoVersion=4.0.2
+analyticsVersion=0.1.0

server/build.gradle

@@ -143,6 +143,10 @@ dependencies {
     implementation("org.eclipse.microprofile.openapi:microprofile-openapi-api:${project.microprofileOpenApiVersion}")
     implementation("jakarta.ws.rs:jakarta.ws.rs-api:${project.jakartaWsRsVersion}")
 
+    implementation(group: "org.apache.cassandra", name: "cassandra-analytics-common", version: "${[project.analyticsVersion]}")
+    implementation(group: "org.apache.cassandra", name: "cassandra-analytics-cdc_spark3_2.12", version: "${[project.analyticsVersion]}")
+    implementation "com.esotericsoftware:kryo-shaded:${kryoVersion}"
+
     testImplementation "org.junit.jupiter:junit-jupiter-api:${project.junitVersion}"
     testImplementation "org.junit.jupiter:junit-jupiter-params:${project.junitVersion}"
     testImplementation "org.assertj:assertj-core:3.24.2"

server/src/main/java/org/apache/cassandra/bridge/CassandraBridgeFactory.java

@@ -0,0 +1,140 @@
+/*
+ * 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.bridge;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+
+
+import jakarta.inject.Singleton;
+import org.jetbrains.annotations.NotNull;
+
+import static org.apache.cassandra.bridge.BaseCassandraBridgeFactory.getCassandraVersion;
+
+/**
+ * Factory class for creating Cassandra bridge instances based on version-specific jar files.
+ * <p>
+ * This factory maintains a cache of CassandraBridge instances mapped by Cassandra version labels
+ * and provides methods to retrieve bridge instances for specific Cassandra versions.
+ * Each bridge is loaded from version-specific JAR resources and instantiated using reflection.
+ */
+@Singleton
+public class CassandraBridgeFactory
+{
+    // maps Cassandra version-specific jar name (e.g. 'four-zero') to matching CassandraBridge
+    private final Map<String, CassandraBridge> cassandraBridges;
+    
+    public CassandraBridgeFactory()
+    {
+        cassandraBridges = new ConcurrentHashMap<>(CassandraVersion.values().length);
+    }
+
+    @NotNull
+    public CassandraBridge get(@NotNull String version)
+    {
+        return get(getCassandraVersion(version));
+    }
+
+    @NotNull
+    public CassandraBridge get(@NotNull CassandraVersionFeatures features)
+    {
+        return get(getCassandraVersion(features));
+    }
+
+    @NotNull
+    public CassandraBridge get(@NotNull CassandraVersion version)
+    {
+        String jarBaseName = version.jarBaseName();
+        Objects.requireNonNull(jarBaseName, "Cassandra version " + version + " is not supported");
+        return cassandraBridges.computeIfAbsent(jarBaseName, this::create);
+    }
+
+    @NotNull
+    @SuppressWarnings("unchecked")
+    private CassandraBridge create(@NotNull String label)
+    {
+        try
+        {
+            ClassLoader loader = buildClassLoader(
+            cassandraResourceName(label),
+            bridgeResourceName(label),
+            typesResourceName(label));
+            Class<CassandraBridge> bridge = (Class<CassandraBridge>) loader.loadClass(CassandraBridge.IMPLEMENTATION_FQCN);
+            Constructor<CassandraBridge> constructor = bridge.getConstructor();
+            return constructor.newInstance();
+        }
+        catch (ClassNotFoundException | NoSuchMethodException | InstantiationException
+               | IllegalAccessException | InvocationTargetException exception)
+        {
+            throw new RuntimeException("Failed to create Cassandra bridge for label " + label, exception);
+        }
+    }
+
+    @NotNull
+    String cassandraResourceName(@NotNull String label)
+    {
+        return "/bridges/" + label + ".jar";
+    }
+
+    @NotNull
+    String bridgeResourceName(@NotNull String label)
+    {
+        return jarResourceName(label, "bridge");
+    }
+
+    @NotNull
+    String typesResourceName(@NotNull String label)
+    {
+        return jarResourceName(label, "types");
+    }
+
+    String jarResourceName(String... parts)
+    {
+        return "/bridges/" + String.join("-", parts) + ".jar";
+    }
+
+    public ClassLoader buildClassLoader(String... resourceNames)
+    {
+        URL[] urls = Arrays.stream(resourceNames)
+                           .map(BaseCassandraBridgeFactory::copyClassResourceToFile)
+                           .map(jar -> {
+                               try
+                               {
+                                   return jar.toURI().toURL();
+                               }
+                               catch (MalformedURLException e)
+                               {
+                                   throw new RuntimeException(e);
+                               }
+                           }).toArray(URL[]::new);
+
+        return AccessController.doPrivileged((PrivilegedAction<ClassLoader>) () ->
+                                                                             new PostDelegationClassLoader(urls, Thread.currentThread().getContextClassLoader()));
+    }
+
+}