This is an automated email from the ASF dual-hosted git repository.
jyothsnakonisa pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-sidecar.git
The following commit(s) were added to refs/heads/trunk by this push:
new 238057afd CASSSIDECAR-507: CDC fails to deserialize nested UDT columns
(#394)
238057afd is described below
commit 238057afdc70d8bc2052df72d3a0fa9b4f7001e9
Author: Jyothsna konisa <[email protected]>
AuthorDate: Fri Oct 16 14:34:39 2026 -0700
CASSSIDECAR-507: CDC fails to deserialize nested UDT columns (#394)
Patch by Jyothsna Konisa; Reviewed by Yifan Cai for CASSSIDECAR-507
---
CHANGES.txt | 1 +
gradle.properties | 2 +
.../sidecar/testing/TestCdcPublisher.java | 4 +-
integration-tests/build.gradle | 2 +
.../cassandra/sidecar/cdc/CdcIntegrationTest.java | 80 +++++++++
...SharedClusterCdcSidecarIntegrationTestBase.java | 3 -
.../sidecar/bridge/CassandraBridgeFactory.java | 124 -------------
.../apache/cassandra/sidecar/cdc/CdcPublisher.java | 7 +-
.../cassandra/sidecar/cdc/CdcSchemaSupplier.java | 7 +-
.../cdc/SidecarCqlToAvroSchemaConverter.java | 8 +-
.../cassandra/sidecar/modules/CdcModule.java | 24 +--
.../tasks/CassandraClusterSchemaMonitor.java | 8 +-
.../sidecar/cdc/CachingSchemaStoreTest.java | 3 +-
.../cassandra/sidecar/cdc/CdcPublisherTests.java | 15 +-
.../sidecar/cdc/CdcSchemaSupplierTest.java | 36 ++--
.../tasks/CassandraClusterSchemaMonitorTest.java | 196 ++++++++-------------
16 files changed, 186 insertions(+), 334 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 28fb6f030..0c2a88f59 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,6 @@
0.5.0
-----
+ * CDC fails to deserialize nested UDT columns (CASSSIDECAR-507)
* CDC log_only mode should not require Kafka producer configuration
(CASSSIDECAR-492)
* Report unknown node status as "Unknown" instead of "?" in ring and token
range replica responses (CASSSIDECAR-506)
* Remove unused Instance Metadata Fetcher from CdcDatabaseAccessor
(CASSSIDECAR-497)
diff --git a/gradle.properties b/gradle.properties
index 7d9c628a9..8ef5942bc 100644
--- a/gradle.properties
+++ b/gradle.properties
@@ -44,6 +44,8 @@ jakartaWsRsVersion=3.1.0
swaggerVersion=2.2.21
# Cdc dependencies
kryoVersion=4.0.2
+# Avro version provided at runtime by the CDC dependencies
(kafka-avro-serializer, Cassandra Analytics)
+avroVersion=1.11.4
# OSHI dependencies
oshiVersion=6.9.0
analyticsVersion=0.5-SNAPSHOT
diff --git
a/integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/TestCdcPublisher.java
b/integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/TestCdcPublisher.java
index 1aca85276..8ac08530c 100644
---
a/integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/TestCdcPublisher.java
+++
b/integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/TestCdcPublisher.java
@@ -27,7 +27,6 @@ import org.apache.cassandra.cdc.sidecar.ClusterConfigProvider;
import org.apache.cassandra.cdc.sidecar.SidecarCdcClient;
import org.apache.cassandra.cdc.sidecar.SidecarCdcStats;
import org.apache.cassandra.cdc.stats.ICdcStats;
-import org.apache.cassandra.sidecar.bridge.CassandraBridgeFactory;
import org.apache.cassandra.sidecar.cdc.CachingSchemaStore;
import org.apache.cassandra.sidecar.cdc.CdcConfig;
import org.apache.cassandra.sidecar.cdc.CdcPublisher;
@@ -61,14 +60,13 @@ public class TestCdcPublisher extends CdcPublisher
CdcSystemViewsDatabaseAccessor cdcSystemViews,
SidecarCdcStats sidecarCdcStats,
Provider<RangeManager> rangeManagerProvider,
- CassandraBridgeFactory cassandraBridgeFactory,
Provider<SidecarCdcClient>
sidecarCdcClientProvider,
CdcOptions cdcOptions)
{
super(vertx, executorPools, clusterConfigProvider,
schemaSupplier, instanceMetadataFetcher, conf, databaseAccessor,
cdcStats,
cdcSystemViews, sidecarCdcStats, rangeManagerProvider,
- cassandraBridgeFactory, sidecarCdcClientProvider,
+ sidecarCdcClientProvider,
mock(CachingSchemaStore.class),
mock(KafkaProducerFactory.class),
cdcOptions);
diff --git a/integration-tests/build.gradle b/integration-tests/build.gradle
index c34a7503f..e044a00b9 100644
--- a/integration-tests/build.gradle
+++ b/integration-tests/build.gradle
@@ -76,6 +76,8 @@ dependencies {
testRuntimeOnly 'com.fasterxml.jackson.core:jackson-annotations:2.14.2'
integrationTestCompileOnly(group: 'com.intellij', name: 'annotations',
version: '12.0')
+ // Avro is provided at runtime by the CDC dependencies; compile only, to
serialize CDC events in tests
+ integrationTestCompileOnly("org.apache.avro:avro:${project.avroVersion}")
}
apply from: "${project.rootDir}/gradle/common/integrationTestTask.gradle"
diff --git
a/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/cdc/CdcIntegrationTest.java
b/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/cdc/CdcIntegrationTest.java
index 3ccf92b79..9584693c3 100644
---
a/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/cdc/CdcIntegrationTest.java
+++
b/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/cdc/CdcIntegrationTest.java
@@ -17,6 +17,8 @@
*/
package org.apache.cassandra.sidecar.cdc;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.List;
@@ -24,15 +26,30 @@ import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import org.junit.jupiter.api.Test;
+import com.google.inject.Injector;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericDatumReader;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.io.BinaryEncoder;
+import org.apache.avro.io.DecoderFactory;
+import org.apache.avro.io.EncoderFactory;
+import org.apache.cassandra.bridge.CassandraVersion;
+import org.apache.cassandra.bridge.CdcBridgeFactory;
+import org.apache.cassandra.cdc.TypeCache;
+import org.apache.cassandra.cdc.avro.AvroConstants;
+import org.apache.cassandra.cdc.avro.AvroGenericRecordTransformer;
import org.apache.cassandra.cdc.msg.CdcEvent;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.sidecar.tasks.CassandraClusterSchemaMonitor;
import org.apache.cassandra.sidecar.testing.QualifiedName;
import
org.apache.cassandra.sidecar.testing.SharedClusterCdcSidecarIntegrationTestBase;
import org.apache.cassandra.sidecar.testing.TestCdcEventConsumer;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
import org.apache.cassandra.spark.utils.TableIdentifier;
import static org.apache.cassandra.testing.TestUtils.DC1_RF1;
@@ -62,6 +79,7 @@ public class CdcIntegrationTest extends
SharedClusterCdcSidecarIntegrationTestBa
private static final QualifiedName CDC_TABLE = new
QualifiedName("cdc_batch_test_ks", "cdc_table");
private static final QualifiedName NON_CDC_TABLE = new
QualifiedName("cdc_batch_test_ks", "non_cdc_table");
private static final QualifiedName MISMATCHED_PK_TABLE = new
QualifiedName("cdc_batch_test_ks", "mismatched_pk_table");
+ private static final QualifiedName UDT_TABLE = new
QualifiedName("cdc_udt_test_ks", "udt_table");
@Override
protected void initializeSchemaForTest()
@@ -82,6 +100,13 @@ public class CdcIntegrationTest extends
SharedClusterCdcSidecarIntegrationTestBa
createTestTable(MISMATCHED_PK_TABLE, "CREATE TABLE IF NOT EXISTS %s "
+ "(id text PRIMARY KEY, value
int) "
+ "WITH cdc = false");
+
+ createTestKeyspace(UDT_TABLE, DC1_RF1);
+ cluster.schemaChangeIgnoringStoppedInstances("CREATE TYPE IF NOT
EXISTS " + UDT_TABLE.keyspace()
+ + ".task_result (status
text, message text)");
+ createTestTable(UDT_TABLE, "CREATE TABLE IF NOT EXISTS %s "
+ + "(id int PRIMARY KEY, results
list<frozen<task_result>>) "
+ + "WITH cdc = true");
}
@Override
@@ -171,6 +196,61 @@ public class CdcIntegrationTest extends
SharedClusterCdcSidecarIntegrationTestBa
.noneMatch(event -> event.table.equals(NON_CDC_TABLE.table()));
}
+ @Test
+ void testUdtColumnsArePublishedAsAvroRecords() throws IOException
+ {
+ String query = String.format("INSERT INTO %s (id, results) VALUES (1,
" +
+ "[{status: 'SUCCESS', message:
'completed'}, {status: 'FAILED', message: 'timed out'}])",
+ UDT_TABLE);
+ cluster.getFirstRunningInstance()
+ .coordinator()
+ .execute(query, ConsistencyLevel.ONE);
+
+ // Seal the active commit log segment so CDC can find the mutation in
cdc_raw
+ cluster.getFirstRunningInstance().flush(UDT_TABLE.keyspace());
+
+ TestCdcEventConsumer consumer = getTestEventConsumer();
+ AtomicReference<CdcEvent> udtEvent = new AtomicReference<>();
+ waitUntil(() -> {
+ consumer.getEvents().stream()
+ .filter(event -> event.table.equals(UDT_TABLE.table()))
+ .findFirst()
+ .ifPresent(udtEvent::set);
+ return udtEvent.get() != null;
+ }, 120, 1000);
+ CdcEvent event = udtEvent.get();
+
+ // Serialize the event to Avro as the Kafka publisher does, with
Sidecar's schema store and type lookup
+ Injector injector = serverWrapper.injector;
+ String releaseVersion =
injector.getInstance(InstanceMetadataFetcher.class)
+ .callOnFirstAvailableInstance(instance
-> instance.delegate().nodeSettings())
+ .releaseVersion();
+ CassandraVersion version =
CdcBridgeFactory.get(releaseVersion).getVersion();
+ AvroGenericRecordTransformer transformer =
+ new
AvroGenericRecordTransformer(injector.getInstance(CachingSchemaStore.class),
+ key ->
TypeCache.get(version).getType(key.keyspace, key.type),
+ null);
+ GenericData.Record record = transformer.transform(event);
+ GenericRecord payload = (GenericRecord)
encodeAndDecode(record).get(AvroConstants.PAYLOAD_KEY);
+
+ List<?> results = (List<?>) payload.get("results");
+ assertThat(results).hasSize(2);
+ assertThat(((GenericRecord)
results.get(0)).get("status")).hasToString("SUCCESS");
+ assertThat(((GenericRecord)
results.get(0)).get("message")).hasToString("completed");
+ assertThat(((GenericRecord)
results.get(1)).get("status")).hasToString("FAILED");
+ assertThat(((GenericRecord)
results.get(1)).get("message")).hasToString("timed out");
+ }
+
+ private static GenericRecord encodeAndDecode(GenericData.Record record)
throws IOException
+ {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(out, null);
+ new
GenericDatumWriter<GenericRecord>(record.getSchema()).write(record, encoder);
+ encoder.flush();
+ return new GenericDatumReader<GenericRecord>(record.getSchema())
+ .read(null,
DecoderFactory.get().binaryDecoder(out.toByteArray(), null));
+ }
+
@Test
void testTableWithMismatchedPartitionKeyIsNeverRegistered()
{
diff --git
a/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/testing/SharedClusterCdcSidecarIntegrationTestBase.java
b/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/testing/SharedClusterCdcSidecarIntegrationTestBase.java
index 135ee6d4b..98c227ed8 100644
---
a/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/testing/SharedClusterCdcSidecarIntegrationTestBase.java
+++
b/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/testing/SharedClusterCdcSidecarIntegrationTestBase.java
@@ -36,7 +36,6 @@ import org.apache.cassandra.cdc.sidecar.SidecarCdcStats;
import org.apache.cassandra.cdc.stats.ICdcStats;
import org.apache.cassandra.distributed.api.ICluster;
import org.apache.cassandra.distributed.api.IInstance;
-import org.apache.cassandra.sidecar.bridge.CassandraBridgeFactory;
import org.apache.cassandra.sidecar.cdc.CdcConfig;
import org.apache.cassandra.sidecar.cdc.CdcPublisher;
import
org.apache.cassandra.sidecar.common.server.utils.MillisecondBoundConfiguration;
@@ -182,7 +181,6 @@ public abstract class
SharedClusterCdcSidecarIntegrationTestBase extends SharedC
CdcSystemViewsDatabaseAccessor systemViews,
SidecarCdcStats sidecarCdcStats,
TokenRingProvider tokenRingProvider,
- CassandraBridgeFactory
cassandraBridgeFactory,
Provider<SidecarCdcClient>
sidecarCdcClientProvider,
CdcOptions cdcOptions)
{
@@ -198,7 +196,6 @@ public abstract class
SharedClusterCdcSidecarIntegrationTestBase extends SharedC
systemViews,
sidecarCdcStats,
() -> rangeManager,
- cassandraBridgeFactory,
sidecarCdcClientProvider,
cdcOptions);
}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/bridge/CassandraBridgeFactory.java
b/server/src/main/java/org/apache/cassandra/sidecar/bridge/CassandraBridgeFactory.java
deleted file mode 100644
index 8dc493500..000000000
---
a/server/src/main/java/org/apache/cassandra/sidecar/bridge/CassandraBridgeFactory.java
+++ /dev/null
@@ -1,124 +0,0 @@
-/*
- * 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.sidecar.bridge;
-
-import java.lang.reflect.Constructor;
-import java.lang.reflect.InvocationTargetException;
-import java.security.AccessController;
-import java.security.PrivilegedAction;
-import java.util.Map;
-import java.util.Objects;
-import java.util.concurrent.ConcurrentHashMap;
-
-import org.apache.cassandra.bridge.BaseCassandraBridgeFactory;
-import org.apache.cassandra.bridge.CassandraBridge;
-import org.apache.cassandra.bridge.CassandraVersion;
-import org.apache.cassandra.bridge.CassandraVersionFeatures;
-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.
- */
-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)
- {
- return AccessController.doPrivileged((PrivilegedAction<ClassLoader>)
() -> BaseCassandraBridgeFactory.buildClassLoader(resourceNames));
- }
-
-}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java
b/server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java
index f52cdf1fa..8981f7bd7 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java
@@ -28,6 +28,7 @@ import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.core.eventbus.Message;
import org.apache.cassandra.bridge.CassandraVersion;
+import org.apache.cassandra.bridge.CdcBridgeFactory;
import org.apache.cassandra.cdc.CdcLogMode;
import org.apache.cassandra.cdc.TypeCache;
import org.apache.cassandra.cdc.api.CdcOptions;
@@ -40,7 +41,6 @@ import org.apache.cassandra.cdc.sidecar.ClusterConfigProvider;
import org.apache.cassandra.cdc.sidecar.SidecarCdcClient;
import org.apache.cassandra.cdc.sidecar.SidecarCdcStats;
import org.apache.cassandra.cdc.stats.ICdcStats;
-import org.apache.cassandra.sidecar.bridge.CassandraBridgeFactory;
import org.apache.cassandra.sidecar.common.server.utils.DurationSpec;
import
org.apache.cassandra.sidecar.common.server.utils.MillisecondBoundConfiguration;
import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
@@ -78,7 +78,6 @@ public class CdcPublisher implements
Handler<Message<Object>>, PeriodicTask
private final ICdcStats cdcStats;
private CdcManager cdcManager;
private final Provider<RangeManager> rangeManagerProvider;
- private final CassandraBridgeFactory cassandraBridgeFactory;
private final CachingSchemaStore schemaStore;
private KafkaPublisher<?> kafkaPublisher;
private final Provider<SidecarCdcClient> sidecarCdcClientProvider;
@@ -97,7 +96,6 @@ public class CdcPublisher implements
Handler<Message<Object>>, PeriodicTask
CdcSystemViewsDatabaseAccessor systemViews,
SidecarCdcStats sidecarCdcStats,
Provider<RangeManager> rangeManagerProvider,
- CassandraBridgeFactory cassandraBridgeFactory,
Provider<SidecarCdcClient> sidecarCdcClientProvider,
CachingSchemaStore schemaStore,
KafkaProducerFactory kafkaProducerFactory,
@@ -113,7 +111,6 @@ public class CdcPublisher implements
Handler<Message<Object>>, PeriodicTask
this.clusterConfigProvider = clusterConfigProvider;
this.cdcStats = cdcStats;
this.rangeManagerProvider = rangeManagerProvider;
- this.cassandraBridgeFactory = cassandraBridgeFactory;
this.sidecarCdcClientProvider = sidecarCdcClientProvider;
this.schemaStore = schemaStore;
this.kafkaProducerFactory = kafkaProducerFactory;
@@ -137,7 +134,7 @@ public class CdcPublisher implements
Handler<Message<Object>>, PeriodicTask
public EventConsumer eventConsumer(CdcConfig conf)
{
- CassandraVersion version = cassandraBridgeFactory.get(
+ CassandraVersion version = CdcBridgeFactory.get(
instanceMetadataFetcher.callOnFirstAvailableInstance(instance ->
instance.delegate().nodeSettings()).releaseVersion()
).getVersion();
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcSchemaSupplier.java
b/server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcSchemaSupplier.java
index 2fe23dd7c..c39d55db1 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcSchemaSupplier.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcSchemaSupplier.java
@@ -24,8 +24,8 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.cassandra.bridge.CassandraBridge;
+import org.apache.cassandra.bridge.CdcBridgeFactory;
import org.apache.cassandra.cdc.api.SchemaSupplier;
-import org.apache.cassandra.sidecar.bridge.CassandraBridgeFactory;
import org.apache.cassandra.sidecar.common.response.NodeSettings;
import org.apache.cassandra.sidecar.config.SidecarConfiguration;
import org.apache.cassandra.sidecar.db.CdcDatabaseAccessor;
@@ -51,18 +51,15 @@ import org.apache.cassandra.spark.utils.TableIdentifier;
public class CdcSchemaSupplier implements SchemaSupplier
{
private final InstanceMetadataFetcher instanceMetadataFetcher;
- private final CassandraBridgeFactory cassandraBridgeFactory;
private final CdcDatabaseAccessor cdcDatabaseAccessor;
private final SidecarConfiguration sidecarConfiguration;
private final ConcurrentHashMap<TableIdentifier, UUID> tableIdCache = new
ConcurrentHashMap<>();
public CdcSchemaSupplier(InstanceMetadataFetcher instanceMetadataFetcher,
- CassandraBridgeFactory cassandraBridgeFactory,
CdcDatabaseAccessor cdcDatabaseAccessor,
SidecarConfiguration sidecarConfiguration)
{
this.instanceMetadataFetcher = instanceMetadataFetcher;
- this.cassandraBridgeFactory = cassandraBridgeFactory;
this.cdcDatabaseAccessor = cdcDatabaseAccessor;
this.sidecarConfiguration = sidecarConfiguration;
}
@@ -79,7 +76,7 @@ public class CdcSchemaSupplier implements SchemaSupplier
instance ->
instance.delegate().metadata().exportSchemaAsString());
NodeSettings nodeSettings =
instanceMetadataFetcher.callOnFirstAvailableInstance(
instance -> instance.delegate().nodeSettings());
- CassandraBridge cassandraBridge =
cassandraBridgeFactory.get(nodeSettings.releaseVersion());
+ CassandraBridge cassandraBridge =
CdcBridgeFactory.get(nodeSettings.releaseVersion());
Partitioner partitioner = getPartitioner(nodeSettings);
boolean batchStatementsEnabled =
sidecarConfiguration.serviceConfiguration().cdcConfiguration().batchStatementsEnabled();
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/cdc/SidecarCqlToAvroSchemaConverter.java
b/server/src/main/java/org/apache/cassandra/sidecar/cdc/SidecarCqlToAvroSchemaConverter.java
index ba8bc03a3..790b8978d 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/cdc/SidecarCqlToAvroSchemaConverter.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/cdc/SidecarCqlToAvroSchemaConverter.java
@@ -24,7 +24,6 @@ import org.apache.avro.Schema;
import org.apache.cassandra.bridge.CassandraBridge;
import org.apache.cassandra.bridge.CdcBridgeFactory;
import org.apache.cassandra.cdc.avro.CqlToAvroSchemaConverter;
-import org.apache.cassandra.sidecar.bridge.CassandraBridgeFactory;
import org.apache.cassandra.sidecar.common.response.NodeSettings;
import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
import org.apache.cassandra.spark.data.CqlTable;
@@ -35,21 +34,18 @@ import org.apache.cassandra.spark.data.CqlTable;
public class SidecarCqlToAvroSchemaConverter implements
CqlToAvroSchemaConverter
{
private final InstanceMetadataFetcher instanceMetadataFetcher;
- private final CassandraBridgeFactory cassandraBridgeFactory;
@Inject
- public SidecarCqlToAvroSchemaConverter(InstanceMetadataFetcher
instanceMetadataFetcher,
- CassandraBridgeFactory
cassandraBridgeFactory)
+ public SidecarCqlToAvroSchemaConverter(InstanceMetadataFetcher
instanceMetadataFetcher)
{
this.instanceMetadataFetcher = instanceMetadataFetcher;
- this.cassandraBridgeFactory = cassandraBridgeFactory;
}
public CassandraBridge cassandraBridge()
{
NodeSettings nodeSettings =
instanceMetadataFetcher.callOnFirstAvailableInstance(instance->
instance.delegate().nodeSettings());
- return cassandraBridgeFactory.get(nodeSettings.releaseVersion());
+ return CdcBridgeFactory.get(nodeSettings.releaseVersion());
}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/modules/CdcModule.java
b/server/src/main/java/org/apache/cassandra/sidecar/modules/CdcModule.java
index 86435bdce..797a897e9 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/modules/CdcModule.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/modules/CdcModule.java
@@ -48,7 +48,6 @@ import org.apache.cassandra.cdc.sidecar.SidecarCdcStats;
import org.apache.cassandra.cdc.stats.CdcStats;
import org.apache.cassandra.cdc.stats.ICdcStats;
import org.apache.cassandra.secrets.SecretsProvider;
-import org.apache.cassandra.sidecar.bridge.CassandraBridgeFactory;
import org.apache.cassandra.sidecar.cdc.CachingSchemaStore;
import org.apache.cassandra.sidecar.cdc.CdcConfig;
import org.apache.cassandra.sidecar.cdc.CdcConfigImpl;
@@ -158,12 +157,10 @@ public class CdcModule extends AbstractModule
CassandraClusterSchemaMonitor
cassandraClusterSchemaMonitorInstance(InstanceMetadataFetcher
instanceMetadataFetcher,
CdcDatabaseAccessor databaseAccessor,
DriverUnsupportedSchemaCache driverUnsupportedSchemaCache,
-
SidecarConfiguration configuration,
-
CassandraBridgeFactory cassandraBridgeFactory)
+
SidecarConfiguration configuration)
{
return new CassandraClusterSchemaMonitor(instanceMetadataFetcher,
databaseAccessor,
- driverUnsupportedSchemaCache,
configuration,
- cassandraBridgeFactory);
+ driverUnsupportedSchemaCache,
configuration);
}
@ProvidesIntoMap
@@ -176,11 +173,10 @@ public class CdcModule extends AbstractModule
@Provides
@Singleton
- CqlToAvroSchemaConverter cqlToAvroSchemaConverter(InstanceMetadataFetcher
instanceMetadataFetcher,
- CassandraBridgeFactory
cassandraBridgeFactory)
+ CqlToAvroSchemaConverter cqlToAvroSchemaConverter(InstanceMetadataFetcher
instanceMetadataFetcher)
{
- return new SidecarCqlToAvroSchemaConverter(instanceMetadataFetcher,
cassandraBridgeFactory);
+ return new SidecarCqlToAvroSchemaConverter(instanceMetadataFetcher);
}
@Provides
@@ -191,13 +187,6 @@ public class CdcModule extends AbstractModule
return new CdcConfigImpl(vertx, cdcConfigAccessor);
}
- @Provides
- @Singleton
- CassandraBridgeFactory cassandraBridgeFactory()
- {
- return new CassandraBridgeFactory();
- }
-
@Provides
@Singleton
SchemaStorePublisherFactory schemaStorePublisherFactory()
@@ -361,11 +350,10 @@ public class CdcModule extends AbstractModule
@Provides
@Singleton
public SchemaSupplier schemaSupplier(InstanceMetadataFetcher
instanceMetadataFetcher,
- CassandraBridgeFactory
cassandraBridgeFactory,
CdcDatabaseAccessor
cdcDatabaseAccessor,
SidecarConfiguration
sidecarConfiguration)
{
- return new CdcSchemaSupplier(instanceMetadataFetcher,
cassandraBridgeFactory, cdcDatabaseAccessor, sidecarConfiguration);
+ return new CdcSchemaSupplier(instanceMetadataFetcher,
cdcDatabaseAccessor, sidecarConfiguration);
}
@Provides
@@ -465,7 +453,6 @@ public class CdcModule extends AbstractModule
CdcSystemViewsDatabaseAccessor systemViews,
SidecarCdcStats sidecarCdcStats,
RangeManager rangeManager,
- CassandraBridgeFactory cassandraBridgeFactory,
Provider<SidecarCdcClient>
sidecarCdcClientProvider,
CachingSchemaStore schemaStore,
KafkaProducerFactory kafkaProducerFactory,
@@ -482,7 +469,6 @@ public class CdcModule extends AbstractModule
systemViews,
sidecarCdcStats,
() -> rangeManager,
- cassandraBridgeFactory,
sidecarCdcClientProvider,
schemaStore,
kafkaProducerFactory,
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/tasks/CassandraClusterSchemaMonitor.java
b/server/src/main/java/org/apache/cassandra/sidecar/tasks/CassandraClusterSchemaMonitor.java
index 1a948fe52..054d08a7d 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/tasks/CassandraClusterSchemaMonitor.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/tasks/CassandraClusterSchemaMonitor.java
@@ -36,7 +36,6 @@ import io.vertx.core.Promise;
import org.apache.cassandra.bridge.CassandraBridge;
import org.apache.cassandra.bridge.CdcBridge;
import org.apache.cassandra.bridge.CdcBridgeFactory;
-import org.apache.cassandra.sidecar.bridge.CassandraBridgeFactory;
import org.apache.cassandra.sidecar.cdc.CdcSchemaUtils;
import org.apache.cassandra.sidecar.common.response.NodeSettings;
import org.apache.cassandra.sidecar.common.server.utils.DurationSpec;
@@ -83,20 +82,17 @@ public class CassandraClusterSchemaMonitor implements
PeriodicTask
private final CopyOnWriteArrayList<Runnable> schemaChangeListeners = new
CopyOnWriteArrayList<>();
private final SidecarConfiguration sidecarConfiguration;
private final InstanceMetadataFetcher instanceFetcher;
- private final CassandraBridgeFactory cassandraBridgeFactory;
public CassandraClusterSchemaMonitor(InstanceMetadataFetcher
instanceFetcher,
CdcDatabaseAccessor databaseAccessor,
DriverUnsupportedSchemaCache
driverUnsupportedSchemaCache,
- SidecarConfiguration
sidecarConfiguration,
- CassandraBridgeFactory
cassandraBridgeFactory)
+ SidecarConfiguration
sidecarConfiguration)
{
this.instanceFetcher = instanceFetcher;
this.databaseAccessor = databaseAccessor;
this.driverUnsupportedSchemaCache = driverUnsupportedSchemaCache;
this.sidecarConfiguration = sidecarConfiguration;
- this.cassandraBridgeFactory = cassandraBridgeFactory;
}
public void addSchemaChangeListener(Runnable listener)
@@ -107,7 +103,7 @@ public class CassandraClusterSchemaMonitor implements
PeriodicTask
public void refresh()
{
NodeSettings nodeSettings =
instanceFetcher.callOnFirstAvailableInstance(instance ->
instance.delegate().nodeSettings());
- CassandraBridge cassandraBridge =
cassandraBridgeFactory.get(nodeSettings.releaseVersion());
+ CassandraBridge cassandraBridge =
CdcBridgeFactory.get(nodeSettings.releaseVersion());
CdcBridge cdcBridge = CdcBridgeFactory.getCdcBridge(cassandraBridge);
try
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/cdc/CachingSchemaStoreTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/cdc/CachingSchemaStoreTest.java
index ecc1c2a22..347cb2678 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/cdc/CachingSchemaStoreTest.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/cdc/CachingSchemaStoreTest.java
@@ -44,7 +44,6 @@ import org.apache.cassandra.cdc.avro.CqlToAvroSchemaConverter;
import org.apache.cassandra.cdc.schemastore.SchemaStorePublisherFactory;
import org.apache.cassandra.cdc.schemastore.TableSchemaPublisher;
import org.apache.cassandra.cdc.sidecar.SidecarCdcStats;
-import org.apache.cassandra.sidecar.bridge.CassandraBridgeFactory;
import org.apache.cassandra.sidecar.db.TableHistoryDatabaseAccessor;
import org.apache.cassandra.sidecar.db.schema.SidecarSchema;
import org.apache.cassandra.sidecar.tasks.CassandraClusterSchemaMonitor;
@@ -105,7 +104,7 @@ public class CachingSchemaStoreTest
mockSidecarSchema = mock(SidecarSchema.class);
when(mockSidecarSchema.isInitialized()).thenReturn(true);
- cassandraBridge = new CassandraBridgeFactory().get(version);
+ cassandraBridge = CdcBridgeFactory.get(version);
cqlToAvroSchemaConverter =
CdcBridgeFactory.getCqlToAvroSchemaConverter(cassandraBridge);
mockCassandraClusterSchemaMonitor =
mock(CassandraClusterSchemaMonitor.class);
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/cdc/CdcPublisherTests.java
b/server/src/test/java/org/apache/cassandra/sidecar/cdc/CdcPublisherTests.java
index bc252960c..fd3b282b8 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/cdc/CdcPublisherTests.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/cdc/CdcPublisherTests.java
@@ -32,8 +32,6 @@ import com.google.inject.Provider;
import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.core.eventbus.Message;
-import org.apache.cassandra.bridge.CassandraBridge;
-import org.apache.cassandra.bridge.CassandraVersion;
import org.apache.cassandra.cdc.api.CdcOptions;
import org.apache.cassandra.cdc.api.EventConsumer;
import org.apache.cassandra.cdc.api.SchemaSupplier;
@@ -43,7 +41,6 @@ import org.apache.cassandra.cdc.sidecar.ClusterConfigProvider;
import org.apache.cassandra.cdc.sidecar.SidecarCdcClient;
import org.apache.cassandra.cdc.sidecar.SidecarCdcStats;
import org.apache.cassandra.cdc.stats.ICdcStats;
-import org.apache.cassandra.sidecar.bridge.CassandraBridgeFactory;
import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
import org.apache.cassandra.sidecar.common.server.ThrowingRunnable;
import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
@@ -63,7 +60,6 @@ import static
org.apache.cassandra.sidecar.server.SidecarServerEvents.ON_CDC_CAC
import static
org.apache.cassandra.sidecar.server.SidecarServerEvents.ON_SERVER_STOP;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
@@ -105,8 +101,6 @@ public class CdcPublisherTests
@Mock
private Provider<RangeManager> rangeManager;
@Mock
- private CassandraBridgeFactory cassandraBridgeFactory;
- @Mock
private SidecarCdcClient sidecarCdcClient;
@Mock
private CdcOptions cdcOptions;
@@ -141,7 +135,6 @@ public class CdcPublisherTests
systemViews,
sidecarCdcStats,
rangeManager,
- cassandraBridgeFactory,
() -> sidecarCdcClient,
schemaStore,
kafkaProducerFactory,
@@ -195,7 +188,7 @@ public class CdcPublisherTests
new CdcPublisher(
vertx, executorPools, clusterConfigProvider, schemaSupplier,
instanceMetadataFetcher,
enabledConfig, databaseAccessor, cdcStats, systemViews,
freshSidecarCdcStats, rangeManager,
- cassandraBridgeFactory, () -> sidecarCdcClient, schemaStore,
kafkaProducerFactory, cdcOptions
+ () -> sidecarCdcClient, schemaStore, kafkaProducerFactory,
cdcOptions
);
verify(freshSidecarCdcStats).captureCdcEnabled();
@@ -212,7 +205,7 @@ public class CdcPublisherTests
new CdcPublisher(
vertx, executorPools, clusterConfigProvider, schemaSupplier,
instanceMetadataFetcher,
disabledConfig, databaseAccessor, cdcStats, systemViews,
freshSidecarCdcStats, rangeManager,
- cassandraBridgeFactory, () -> sidecarCdcClient, schemaStore,
kafkaProducerFactory, cdcOptions
+ () -> sidecarCdcClient, schemaStore, kafkaProducerFactory,
cdcOptions
);
verify(freshSidecarCdcStats).captureCdcDisabled();
@@ -279,10 +272,6 @@ public class CdcPublisherTests
Function<InstanceMetadata, Object> fn = invocation.getArgument(0);
return fn.apply(mockInstance);
}).when(instanceMetadataFetcher).callOnFirstAvailableInstance(any());
-
- CassandraBridge mockBridge = mock(CassandraBridge.class);
- when(mockBridge.getVersion()).thenReturn(CassandraVersion.FOURONE);
- when(cassandraBridgeFactory.get(anyString())).thenReturn(mockBridge);
}
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/cdc/CdcSchemaSupplierTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/cdc/CdcSchemaSupplierTest.java
index ddb84749b..8d41b019f 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/cdc/CdcSchemaSupplierTest.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/cdc/CdcSchemaSupplierTest.java
@@ -27,9 +27,8 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.apache.cassandra.bridge.CassandraBridge;
-import org.apache.cassandra.bridge.CassandraVersion;
import org.apache.cassandra.bridge.CdcBridge;
-import org.apache.cassandra.sidecar.bridge.CassandraBridgeFactory;
+import org.apache.cassandra.bridge.CdcBridgeFactory;
import org.apache.cassandra.sidecar.common.response.NodeSettings;
import org.apache.cassandra.sidecar.config.CdcConfiguration;
import org.apache.cassandra.sidecar.config.ServiceConfiguration;
@@ -40,13 +39,14 @@ import org.apache.cassandra.spark.data.CqlTable;
import org.apache.cassandra.spark.data.partitioner.Partitioner;
import org.apache.cassandra.spark.utils.TableIdentifier;
import org.mockito.Mock;
+import org.mockito.MockedStatic;
import org.mockito.MockitoAnnotations;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.when;
/**
@@ -57,8 +57,6 @@ public class CdcSchemaSupplierTest
@Mock
private InstanceMetadataFetcher instanceMetadataFetcher;
@Mock
- private CassandraBridgeFactory cassandraBridgeFactory;
- @Mock
private CdcDatabaseAccessor cdcDatabaseAccessor;
@Mock
private CassandraBridge cassandraBridge;
@@ -99,7 +97,6 @@ public class CdcSchemaSupplierTest
cdcSchemaSupplier = new CdcSchemaSupplier(
instanceMetadataFetcher,
- cassandraBridgeFactory,
cdcDatabaseAccessor,
sidecarConfiguration
);
@@ -114,8 +111,6 @@ public class CdcSchemaSupplierTest
.thenReturn(SAMPLE_SCHEMA) // First call returns schema
.thenReturn(nodeSettings); // Second call returns node settings
- CassandraBridge realBridge = new
CassandraBridgeFactory().get(CassandraVersion.FOURONE);
- when(cassandraBridgeFactory.get(anyString())).thenReturn(realBridge);
when(cdcDatabaseAccessor.partitioner()).thenReturn(Partitioner.Murmur3Partitioner);
when(cdcDatabaseAccessor.getTableId(any(TableIdentifier.class))).thenReturn(UUID.randomUUID());
@@ -138,8 +133,6 @@ public class CdcSchemaSupplierTest
.thenReturn(SAMPLE_SCHEMA)
.thenReturn(nodeSettings);
- CassandraBridge realBridge = new
CassandraBridgeFactory().get(CassandraVersion.FOURONE);
- when(cassandraBridgeFactory.get(anyString())).thenReturn(realBridge);
when(cdcDatabaseAccessor.partitioner()).thenReturn(Partitioner.Murmur3Partitioner);
when(cdcDatabaseAccessor.getTableId(any(TableIdentifier.class))).thenReturn(UUID.randomUUID());
@@ -159,8 +152,6 @@ public class CdcSchemaSupplierTest
.thenReturn(SCHEMA_NO_CDC) // Schema with no CDC tables
.thenReturn(nodeSettings);
- CassandraBridge realBridge = new
CassandraBridgeFactory().get(CassandraVersion.FOURONE);
- when(cassandraBridgeFactory.get(anyString())).thenReturn(realBridge);
when(cdcDatabaseAccessor.partitioner()).thenReturn(Partitioner.Murmur3Partitioner);
when(cdcDatabaseAccessor.getTableId(any(TableIdentifier.class))).thenReturn(UUID.randomUUID());
// setUp() already configures batchStatementsEnabled=true — with no
CDC-enabled table in
@@ -182,8 +173,6 @@ public class CdcSchemaSupplierTest
.thenReturn(SCHEMA_NO_CDC) // Keyspace has zero CDC-enabled tables
.thenReturn(nodeSettings);
- CassandraBridge realBridge = new
CassandraBridgeFactory().get(CassandraVersion.FOURONE);
- when(cassandraBridgeFactory.get(anyString())).thenReturn(realBridge);
when(cdcDatabaseAccessor.partitioner()).thenReturn(Partitioner.Murmur3Partitioner);
when(cdcDatabaseAccessor.getTableId(any(TableIdentifier.class))).thenReturn(UUID.randomUUID());
@@ -206,8 +195,6 @@ public class CdcSchemaSupplierTest
.thenReturn(mixedSchema)
.thenReturn(nodeSettings);
- CassandraBridge realBridge = new
CassandraBridgeFactory().get(CassandraVersion.FOURONE);
- when(cassandraBridgeFactory.get(anyString())).thenReturn(realBridge);
when(cdcDatabaseAccessor.partitioner()).thenReturn(Partitioner.Murmur3Partitioner);
// A distinct UUID per table — a single fixed UUID for all tables
collides in the
// bridge's schema-building code once more than one table is built in
the same call.
@@ -235,8 +222,6 @@ public class CdcSchemaSupplierTest
.thenReturn(mixedSchema)
.thenReturn(nodeSettings);
- CassandraBridge realBridge = new
CassandraBridgeFactory().get(CassandraVersion.FOURONE);
- when(cassandraBridgeFactory.get(anyString())).thenReturn(realBridge);
when(cdcDatabaseAccessor.partitioner()).thenReturn(Partitioner.Murmur3Partitioner);
when(cdcDatabaseAccessor.getTableId(any(TableIdentifier.class))).thenAnswer(invocation
-> UUID.randomUUID());
@@ -262,8 +247,6 @@ public class CdcSchemaSupplierTest
.thenReturn(mixedSchema)
.thenReturn(nodeSettings);
- CassandraBridge realBridge = new
CassandraBridgeFactory().get(CassandraVersion.FOURONE);
- when(cassandraBridgeFactory.get(anyString())).thenReturn(realBridge);
when(cdcDatabaseAccessor.partitioner()).thenReturn(Partitioner.Murmur3Partitioner);
when(cdcDatabaseAccessor.getTableId(any(TableIdentifier.class))).thenAnswer(invocation
-> UUID.randomUUID());
@@ -276,7 +259,7 @@ public class CdcSchemaSupplierTest
}
@Test
- void testGetCdcEnabledTablesCallsCassandraBridgeFactory()
+ void testGetCdcEnabledTablesCallsCdcBridgeFactory()
{
String releaseVersion = "4.1.0";
NodeSettings nodeSettings = mockNodeSettings(releaseVersion,
"Murmur3Partitioner");
@@ -285,14 +268,15 @@ public class CdcSchemaSupplierTest
.thenReturn(SAMPLE_SCHEMA)
.thenReturn(nodeSettings);
- CassandraBridge realBridge = new
CassandraBridgeFactory().get(CassandraVersion.FOURONE);
- when(cassandraBridgeFactory.get(anyString())).thenReturn(realBridge);
when(cdcDatabaseAccessor.partitioner()).thenReturn(Partitioner.Murmur3Partitioner);
when(cdcDatabaseAccessor.getTableId(any(TableIdentifier.class))).thenReturn(UUID.randomUUID());
- cdcSchemaSupplier.getTables();
+ try (MockedStatic<CdcBridgeFactory> cdcBridgeFactory =
mockStatic(CdcBridgeFactory.class, CALLS_REAL_METHODS))
+ {
+ cdcSchemaSupplier.getTables();
- verify(cassandraBridgeFactory).get(releaseVersion);
+ cdcBridgeFactory.verify(() ->
CdcBridgeFactory.get(releaseVersion));
+ }
}
private NodeSettings mockNodeSettings(String releaseVersion, String
partitioner)
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/tasks/CassandraClusterSchemaMonitorTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/tasks/CassandraClusterSchemaMonitorTest.java
index 87fc42616..f6e115bcf 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/tasks/CassandraClusterSchemaMonitorTest.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/tasks/CassandraClusterSchemaMonitorTest.java
@@ -26,6 +26,7 @@ import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -33,7 +34,6 @@ import io.vertx.core.Promise;
import org.apache.cassandra.bridge.CassandraBridge;
import org.apache.cassandra.bridge.CdcBridge;
import org.apache.cassandra.bridge.CdcBridgeFactory;
-import org.apache.cassandra.sidecar.bridge.CassandraBridgeFactory;
import org.apache.cassandra.sidecar.common.response.NodeSettings;
import
org.apache.cassandra.sidecar.common.server.utils.SecondBoundConfiguration;
import org.apache.cassandra.sidecar.config.CdcConfiguration;
@@ -80,7 +80,7 @@ class CassandraClusterSchemaMonitorTest
private NodeSettings mockNodeSettings;
private CassandraBridge mockCassandraBridge;
private CdcBridge mockCdcBridge;
- private CassandraBridgeFactory mockCassandraBridgeFactory;
+ private MockedStatic<CdcBridgeFactory> mockCdcBridgeFactory;
private static final String INITIAL_SCHEMA = "CREATE TABLE test.cdc_table
(\n" +
" id uuid PRIMARY KEY,\n" +
@@ -110,7 +110,7 @@ class CassandraClusterSchemaMonitorTest
mockNodeSettings = mock(NodeSettings.class);
mockCassandraBridge = mock(CassandraBridge.class);
mockCdcBridge = mock(CdcBridge.class);
- mockCassandraBridgeFactory = mock(CassandraBridgeFactory.class);
+ mockCdcBridgeFactory = Mockito.mockStatic(CdcBridgeFactory.class);
// Setup configuration chain
when(mockSidecarConfiguration.serviceConfiguration()).thenReturn(mockServiceConfiguration);
@@ -134,17 +134,23 @@ class CassandraClusterSchemaMonitorTest
when(mockDriverUnsupportedSchemaCache.getFullSchema()).thenReturn("");
when(mockDatabaseAccessor.partitioner()).thenReturn(Partitioner.Murmur3Partitioner);
when(mockDatabaseAccessor.getTableId(any(TableIdentifier.class))).thenReturn(UUID.randomUUID());
-
when(mockCassandraBridgeFactory.get(anyString())).thenReturn(mockCassandraBridge);
+ mockCdcBridgeFactory.when(() ->
CdcBridgeFactory.get(anyString())).thenReturn(mockCassandraBridge);
+ mockCdcBridgeFactory.when(() ->
CdcBridgeFactory.getCdcBridge(any(CassandraBridge.class))).thenReturn(mockCdcBridge);
clusterSchema = new CassandraClusterSchemaMonitor(
mockInstanceFetcher,
mockDatabaseAccessor,
mockDriverUnsupportedSchemaCache,
- mockSidecarConfiguration,
- mockCassandraBridgeFactory
+ mockSidecarConfiguration
);
}
+ @AfterEach
+ void tearDown()
+ {
+ mockCdcBridgeFactory.close();
+ }
+
@Test
void testScheduleDecisionExecuteWhenBothConfigurationsEnabled()
{
@@ -198,12 +204,9 @@ class CassandraClusterSchemaMonitorTest
@Test
void testRefreshDetectsSchemaChangeAndUpdatesCdcTables()
{
- try (MockedStatic<CdcBridgeFactory> cdcBridgeFactory =
Mockito.mockStatic(CdcBridgeFactory.class);
- MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
+ try (MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
MockedStatic<CqlUtils> cqlUtils =
Mockito.mockStatic(CqlUtils.class))
{
-
when(mockCassandraBridgeFactory.get(anyString())).thenReturn(mockCassandraBridge);
- cdcBridgeFactory.when(() ->
CdcBridgeFactory.getCdcBridge(any(CassandraBridge.class))).thenReturn(mockCdcBridge);
// Mock utility class calls for initial schema
Map<TableIdentifier, CdcUtil.TableSchema> mockCreateStmts1 =
Collections.singletonMap(
@@ -253,12 +256,9 @@ class CassandraClusterSchemaMonitorTest
@Test
void testRefreshUnregistersTableNoLongerAtRisk()
{
- try (MockedStatic<CdcBridgeFactory> cdcBridgeFactory =
Mockito.mockStatic(CdcBridgeFactory.class);
- MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
+ try (MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
MockedStatic<CqlUtils> cqlUtils =
Mockito.mockStatic(CqlUtils.class))
{
-
when(mockCassandraBridgeFactory.get(anyString())).thenReturn(mockCassandraBridge);
- cdcBridgeFactory.when(() ->
CdcBridgeFactory.getCdcBridge(any(CassandraBridge.class))).thenReturn(mockCdcBridge);
// Initial schema: a CDC table and a non-CDC table sharing the
same partition key
// structure — the non-CDC table is at risk and must be registered.
@@ -307,12 +307,9 @@ class CassandraClusterSchemaMonitorTest
@Test
void testRefreshDoesNotUnregisterWhenNoTableBecomesStale()
{
- try (MockedStatic<CdcBridgeFactory> cdcBridgeFactory =
Mockito.mockStatic(CdcBridgeFactory.class);
- MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
+ try (MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
MockedStatic<CqlUtils> cqlUtils =
Mockito.mockStatic(CqlUtils.class))
{
-
when(mockCassandraBridgeFactory.get(anyString())).thenReturn(mockCassandraBridge);
- cdcBridgeFactory.when(() ->
CdcBridgeFactory.getCdcBridge(any(CassandraBridge.class))).thenReturn(mockCdcBridge);
Map<TableIdentifier, CdcUtil.TableSchema> mockCreateStmts =
Collections.singletonMap(
TableIdentifier.of("test", "cdc_table"), new
CdcUtil.TableSchema(INITIAL_SCHEMA, true,
CdcUtil.PartitionKeySignature.indeterminate()));
@@ -337,12 +334,9 @@ class CassandraClusterSchemaMonitorTest
@Test
void testRefreshRetriesUnregistrationAfterAFailedAttempt()
{
- try (MockedStatic<CdcBridgeFactory> cdcBridgeFactory =
Mockito.mockStatic(CdcBridgeFactory.class);
- MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
+ try (MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
MockedStatic<CqlUtils> cqlUtils =
Mockito.mockStatic(CqlUtils.class))
{
-
when(mockCassandraBridgeFactory.get(anyString())).thenReturn(mockCassandraBridge);
- cdcBridgeFactory.when(() ->
CdcBridgeFactory.getCdcBridge(any(CassandraBridge.class))).thenReturn(mockCdcBridge);
// Same setup as testRefreshUnregistersTableNoLongerAtRisk:
non_cdc_table starts at
// risk (registered), then a schema change makes it no longer at
risk.
@@ -394,7 +388,6 @@ class CassandraClusterSchemaMonitorTest
// the failed unregistration must not have been forgotten,
non_cdc_table is still
// treated as stale and unregisterNonCdcTables is retried with the
same set.
Mockito.reset(mockCdcBridge);
- cdcBridgeFactory.when(() ->
CdcBridgeFactory.getCdcBridge(any(CassandraBridge.class))).thenReturn(mockCdcBridge);
clusterSchema.refresh();
verify(mockCdcBridge,
times(1)).unregisterNonCdcTables(eq(staleSet));
}
@@ -403,12 +396,9 @@ class CassandraClusterSchemaMonitorTest
@Test
void testRefreshSkipsProcessingWhenSchemaUnchanged()
{
- try (MockedStatic<CdcBridgeFactory> cdcBridgeFactory =
Mockito.mockStatic(CdcBridgeFactory.class);
- MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
+ try (MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
MockedStatic<CqlUtils> cqlUtils =
Mockito.mockStatic(CqlUtils.class))
{
-
when(mockCassandraBridgeFactory.get(anyString())).thenReturn(mockCassandraBridge);
- cdcBridgeFactory.when(() ->
CdcBridgeFactory.getCdcBridge(any(CassandraBridge.class))).thenReturn(mockCdcBridge);
// Mock utility class calls
Map<TableIdentifier, CdcUtil.TableSchema> mockCreateStmts =
Collections.singletonMap(
@@ -434,12 +424,9 @@ class CassandraClusterSchemaMonitorTest
@Test
void testRefreshNotifiesSchemaChangeListeners()
{
- try (MockedStatic<CdcBridgeFactory> cdcBridgeFactory =
Mockito.mockStatic(CdcBridgeFactory.class);
- MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
+ try (MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
MockedStatic<CqlUtils> cqlUtils =
Mockito.mockStatic(CqlUtils.class))
{
-
when(mockCassandraBridgeFactory.get(anyString())).thenReturn(mockCassandraBridge);
- cdcBridgeFactory.when(() ->
CdcBridgeFactory.getCdcBridge(any(CassandraBridge.class))).thenReturn(mockCdcBridge);
// Mock utility class calls
Map<TableIdentifier, CdcUtil.TableSchema> mockCreateStmts =
Collections.singletonMap(
@@ -465,64 +452,51 @@ class CassandraClusterSchemaMonitorTest
@Test
void testRefreshHandlesIllegalStateExceptionGracefully()
{
- try (MockedStatic<CdcBridgeFactory> cdcBridgeFactory =
Mockito.mockStatic(CdcBridgeFactory.class))
- {
-
when(mockCassandraBridgeFactory.get(anyString())).thenReturn(mockCassandraBridge);
- cdcBridgeFactory.when(() ->
CdcBridgeFactory.getCdcBridge(any(CassandraBridge.class))).thenReturn(mockCdcBridge);
-
- when(mockDatabaseAccessor.fullSchema()).thenThrow(new
IllegalStateException("Database not ready"));
-
- // Should now throw the exception instead of handling it gracefully
- try
- {
- clusterSchema.refresh();
- assertThat(false).as("Expected IllegalStateException to be
thrown").isTrue();
- }
- catch (IllegalStateException e)
- {
- assertThat(e.getMessage()).isEqualTo("Database not ready");
- }
- verify(mockDatabaseAccessor, times(1)).fullSchema();
- // CdcBridge should not be called due to exception
+ when(mockDatabaseAccessor.fullSchema()).thenThrow(new
IllegalStateException("Database not ready"));
+
+ // Should now throw the exception instead of handling it gracefully
+ try
+ {
+ clusterSchema.refresh();
+ assertThat(false).as("Expected IllegalStateException to be
thrown").isTrue();
}
+ catch (IllegalStateException e)
+ {
+ assertThat(e.getMessage()).isEqualTo("Database not ready");
+ }
+
+ verify(mockDatabaseAccessor, times(1)).fullSchema();
+ // CdcBridge should not be called due to exception
}
@Test
void testRefreshHandlesGenericExceptionGracefully()
{
- try (MockedStatic<CdcBridgeFactory> cdcBridgeFactory =
Mockito.mockStatic(CdcBridgeFactory.class))
- {
-
when(mockCassandraBridgeFactory.get(anyString())).thenReturn(mockCassandraBridge);
- cdcBridgeFactory.when(() ->
CdcBridgeFactory.getCdcBridge(any(CassandraBridge.class))).thenReturn(mockCdcBridge);
-
- when(mockDatabaseAccessor.fullSchema()).thenThrow(new
RuntimeException("Unexpected error"));
-
- // Should now throw the exception instead of handling it gracefully
- try
- {
- clusterSchema.refresh();
- assertThat(false).as("Expected RuntimeException to be
thrown").isTrue();
- }
- catch (RuntimeException e)
- {
- assertThat(e.getMessage()).isEqualTo("Unexpected error");
- }
- verify(mockDatabaseAccessor, times(1)).fullSchema();
- // CdcBridge should not be called due to exception
+ when(mockDatabaseAccessor.fullSchema()).thenThrow(new
RuntimeException("Unexpected error"));
+
+ // Should now throw the exception instead of handling it gracefully
+ try
+ {
+ clusterSchema.refresh();
+ assertThat(false).as("Expected RuntimeException to be
thrown").isTrue();
+ }
+ catch (RuntimeException e)
+ {
+ assertThat(e.getMessage()).isEqualTo("Unexpected error");
}
+
+ verify(mockDatabaseAccessor, times(1)).fullSchema();
+ // CdcBridge should not be called due to exception
}
@Test
void testExecuteCompletesPromiseSuccessfully()
{
- try (MockedStatic<CdcBridgeFactory> cdcBridgeFactory =
Mockito.mockStatic(CdcBridgeFactory.class);
- MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
+ try (MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
MockedStatic<CqlUtils> cqlUtils =
Mockito.mockStatic(CqlUtils.class))
{
-
when(mockCassandraBridgeFactory.get(anyString())).thenReturn(mockCassandraBridge);
- cdcBridgeFactory.when(() ->
CdcBridgeFactory.getCdcBridge(any(CassandraBridge.class))).thenReturn(mockCdcBridge);
// Mock utility class calls
Map<TableIdentifier, CdcUtil.TableSchema> mockCreateStmts =
Collections.singletonMap(
@@ -545,36 +519,28 @@ class CassandraClusterSchemaMonitorTest
@Test
void testExecuteFailsPromiseOnException()
{
- try (MockedStatic<CdcBridgeFactory> cdcBridgeFactory =
Mockito.mockStatic(CdcBridgeFactory.class))
- {
-
when(mockCassandraBridgeFactory.get(anyString())).thenReturn(mockCassandraBridge);
- cdcBridgeFactory.when(() ->
CdcBridgeFactory.getCdcBridge(any(CassandraBridge.class))).thenReturn(mockCdcBridge);
- @SuppressWarnings("unchecked")
- Promise<Void> promise = mock(Promise.class);
- RuntimeException expectedException = new RuntimeException("Refresh
failed");
+ @SuppressWarnings("unchecked")
+ Promise<Void> promise = mock(Promise.class);
+ RuntimeException expectedException = new RuntimeException("Refresh
failed");
-
when(mockDatabaseAccessor.fullSchema()).thenThrow(expectedException);
+ when(mockDatabaseAccessor.fullSchema()).thenThrow(expectedException);
- clusterSchema.execute(promise);
+ clusterSchema.execute(promise);
- ArgumentCaptor<Throwable> throwableCaptor =
ArgumentCaptor.forClass(Throwable.class);
- verify(promise, times(1)).fail(throwableCaptor.capture());
- verify(promise, never()).tryComplete();
+ ArgumentCaptor<Throwable> throwableCaptor =
ArgumentCaptor.forClass(Throwable.class);
+ verify(promise, times(1)).fail(throwableCaptor.capture());
+ verify(promise, never()).tryComplete();
-
assertThat(throwableCaptor.getValue()).isEqualTo(expectedException);
- }
+ assertThat(throwableCaptor.getValue()).isEqualTo(expectedException);
}
@Test
void testAddSchemaChangeListenerStoresListener()
{
- try (MockedStatic<CdcBridgeFactory> cdcBridgeFactory =
Mockito.mockStatic(CdcBridgeFactory.class);
- MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
+ try (MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
MockedStatic<CqlUtils> cqlUtils =
Mockito.mockStatic(CqlUtils.class))
{
-
when(mockCassandraBridgeFactory.get(anyString())).thenReturn(mockCassandraBridge);
- cdcBridgeFactory.when(() ->
CdcBridgeFactory.getCdcBridge(any(CassandraBridge.class))).thenReturn(mockCdcBridge);
// Mock utility class calls
Map<TableIdentifier, CdcUtil.TableSchema> mockCreateStmts =
Collections.singletonMap(
@@ -597,12 +563,9 @@ class CassandraClusterSchemaMonitorTest
@Test
void testMultipleSchemaChangeListenersAllNotified()
{
- try (MockedStatic<CdcBridgeFactory> cdcBridgeFactory =
Mockito.mockStatic(CdcBridgeFactory.class);
- MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
+ try (MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
MockedStatic<CqlUtils> cqlUtils =
Mockito.mockStatic(CqlUtils.class))
{
-
when(mockCassandraBridgeFactory.get(anyString())).thenReturn(mockCassandraBridge);
- cdcBridgeFactory.when(() ->
CdcBridgeFactory.getCdcBridge(any(CassandraBridge.class))).thenReturn(mockCdcBridge);
// Mock utility class calls
Map<TableIdentifier, CdcUtil.TableSchema> mockCreateStmts =
Collections.singletonMap(
@@ -631,12 +594,9 @@ class CassandraClusterSchemaMonitorTest
@Test
void testSchemaChangeListenersNotCalledWhenNoSchemaChange()
{
- try (MockedStatic<CdcBridgeFactory> cdcBridgeFactory =
Mockito.mockStatic(CdcBridgeFactory.class);
- MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
+ try (MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
MockedStatic<CqlUtils> cqlUtils =
Mockito.mockStatic(CqlUtils.class))
{
-
when(mockCassandraBridgeFactory.get(anyString())).thenReturn(mockCassandraBridge);
- cdcBridgeFactory.when(() ->
CdcBridgeFactory.getCdcBridge(any(CassandraBridge.class))).thenReturn(mockCdcBridge);
// Mock utility class calls
Map<TableIdentifier, CdcUtil.TableSchema> mockCreateStmts =
Collections.singletonMap(
@@ -663,40 +623,32 @@ class CassandraClusterSchemaMonitorTest
@Test
void testSchemaChangeListenersNotCalledOnException()
{
- try (MockedStatic<CdcBridgeFactory> cdcBridgeFactory =
Mockito.mockStatic(CdcBridgeFactory.class))
- {
-
when(mockCassandraBridgeFactory.get(anyString())).thenReturn(mockCassandraBridge);
- cdcBridgeFactory.when(() ->
CdcBridgeFactory.getCdcBridge(any(CassandraBridge.class))).thenReturn(mockCdcBridge);
- AtomicBoolean listenerCalled = new AtomicBoolean(false);
- clusterSchema.addSchemaChangeListener(() ->
listenerCalled.set(true));
-
- when(mockDatabaseAccessor.fullSchema()).thenThrow(new
RuntimeException("Error"));
+ AtomicBoolean listenerCalled = new AtomicBoolean(false);
+ clusterSchema.addSchemaChangeListener(() -> listenerCalled.set(true));
- // Should now throw the exception instead of handling it gracefully
- try
- {
- clusterSchema.refresh();
- assertThat(false).as("Expected RuntimeException to be
thrown").isTrue();
- }
- catch (RuntimeException e)
- {
- assertThat(e.getMessage()).isEqualTo("Error");
- }
+ when(mockDatabaseAccessor.fullSchema()).thenThrow(new
RuntimeException("Error"));
- assertThat(listenerCalled.get()).isFalse();
+ // Should now throw the exception instead of handling it gracefully
+ try
+ {
+ clusterSchema.refresh();
+ assertThat(false).as("Expected RuntimeException to be
thrown").isTrue();
+ }
+ catch (RuntimeException e)
+ {
+ assertThat(e.getMessage()).isEqualTo("Error");
}
+
+ assertThat(listenerCalled.get()).isFalse();
}
@Test
void testRefreshUpdatesTableIdCache()
{
- try (MockedStatic<CdcBridgeFactory> cdcBridgeFactory =
Mockito.mockStatic(CdcBridgeFactory.class);
- MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
+ try (MockedStatic<CdcUtil> cdcUtil = Mockito.mockStatic(CdcUtil.class);
MockedStatic<CqlUtils> cqlUtils =
Mockito.mockStatic(CqlUtils.class))
{
-
when(mockCassandraBridgeFactory.get(anyString())).thenReturn(mockCassandraBridge);
- cdcBridgeFactory.when(() ->
CdcBridgeFactory.getCdcBridge(any(CassandraBridge.class))).thenReturn(mockCdcBridge);
// Mock utility class calls
Map<TableIdentifier, CdcUtil.TableSchema> mockCreateStmts =
Collections.singletonMap(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
