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 181bae036 CASSSIDECAR-511: Make the CDC CassandraBridge injectable
(#396)
181bae036 is described below
commit 181bae036fce7a7418beb5a0740f46bed12d4749
Author: Jyothsna konisa <[email protected]>
AuthorDate: Fri Oct 23 17:40:12 2026 -0700
CASSSIDECAR-511: Make the CDC CassandraBridge injectable (#396)
Patch by Jyothsna Konisa; Reviewed by Yifan Cai for CASSSIDECAR-511
---
CHANGES.txt | 1 +
.../sidecar/testing/TestCdcPublisher.java | 4 +-
.../cassandra/sidecar/cdc/CdcIntegrationTest.java | 23 ++--
...SharedClusterCdcSidecarIntegrationTestBase.java | 3 +
.../apache/cassandra/sidecar/cdc/CdcPublisher.java | 10 +-
.../cassandra/sidecar/cdc/CdcSchemaSupplier.java | 7 +-
.../cassandra/sidecar/cdc/SidecarCdcOptions.java | 25 +++-
.../cdc/SidecarCqlToAvroSchemaConverter.java | 12 +-
.../cassandra/sidecar/modules/CdcModule.java | 37 ++++--
.../tasks/CassandraClusterSchemaMonitor.java | 8 +-
.../cassandra/sidecar/cdc/CdcPublisherTests.java | 13 ++-
.../sidecar/cdc/CdcSchemaSupplierTest.java | 20 ++--
.../sidecar/cdc/SidecarCdcOptionsTest.java | 61 +++++++++-
.../modules/CdcModuleCassandraBridgeTest.java | 127 +++++++++++++++++++++
.../tasks/CassandraClusterSchemaMonitorTest.java | 4 +-
15 files changed, 295 insertions(+), 60 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 26d92c8a3..bebe79cd7 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,6 @@
0.5.0
-----
+ * Make the CDC CassandraBridge injectable (CASSSIDECAR-511)
* Introduce server-side configuration to enforce allowed credential modes for
restore jobs (CASSSIDECAR-501)
* Sanitize/Nullify Cloud Credentials on RestoreJob Completion
(CASSSIDECAR-500)
* CDC fails to deserialize nested UDT columns (CASSSIDECAR-507)
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 8ac08530c..dd949e342 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
@@ -19,6 +19,7 @@ package org.apache.cassandra.sidecar.testing;
import com.google.inject.Provider;
import io.vertx.core.Vertx;
+import org.apache.cassandra.bridge.CassandraBridge;
import org.apache.cassandra.cdc.api.CdcOptions;
import org.apache.cassandra.cdc.api.EventConsumer;
import org.apache.cassandra.cdc.api.SchemaSupplier;
@@ -60,13 +61,14 @@ public class TestCdcPublisher extends CdcPublisher
CdcSystemViewsDatabaseAccessor cdcSystemViews,
SidecarCdcStats sidecarCdcStats,
Provider<RangeManager> rangeManagerProvider,
+ Provider<CassandraBridge> cassandraBridgeProvider,
Provider<SidecarCdcClient>
sidecarCdcClientProvider,
CdcOptions cdcOptions)
{
super(vertx, executorPools, clusterConfigProvider,
schemaSupplier, instanceMetadataFetcher, conf, databaseAccessor,
cdcStats,
cdcSystemViews, sidecarCdcStats, rangeManagerProvider,
- sidecarCdcClientProvider,
+ cassandraBridgeProvider, sidecarCdcClientProvider,
mock(CachingSchemaStore.class),
mock(KafkaProducerFactory.class),
cdcOptions);
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 9584693c3..815869ccd 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
@@ -26,7 +26,6 @@ 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;
@@ -38,8 +37,8 @@ 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.CassandraBridge;
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;
@@ -49,7 +48,6 @@ 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;
@@ -210,22 +208,15 @@ public class CdcIntegrationTest extends
SharedClusterCdcSidecarIntegrationTestBa
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();
+ waitUntil(() -> consumer.getEvents().stream().anyMatch(event ->
event.table.equals(UDT_TABLE.table())), 120, 1000);
+ CdcEvent event = consumer.getEvents().stream()
+ .filter(e ->
e.table.equals(UDT_TABLE.table()))
+ .findFirst()
+ .orElseThrow(AssertionError::new);
// 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();
+ CassandraVersion version =
injector.getInstance(CassandraBridge.class).getVersion();
AvroGenericRecordTransformer transformer =
new
AvroGenericRecordTransformer(injector.getInstance(CachingSchemaStore.class),
key ->
TypeCache.get(version).getType(key.keyspace, key.type),
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 98c227ed8..bb1465206 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
@@ -28,6 +28,7 @@ import com.google.inject.Provider;
import com.google.inject.Provides;
import com.google.inject.Singleton;
import io.vertx.core.Vertx;
+import org.apache.cassandra.bridge.CassandraBridge;
import org.apache.cassandra.cdc.api.CdcOptions;
import org.apache.cassandra.cdc.api.SchemaSupplier;
import org.apache.cassandra.cdc.sidecar.ClusterConfigProvider;
@@ -181,6 +182,7 @@ public abstract class
SharedClusterCdcSidecarIntegrationTestBase extends SharedC
CdcSystemViewsDatabaseAccessor systemViews,
SidecarCdcStats sidecarCdcStats,
TokenRingProvider tokenRingProvider,
+ Provider<CassandraBridge>
cassandraBridgeProvider,
Provider<SidecarCdcClient>
sidecarCdcClientProvider,
CdcOptions cdcOptions)
{
@@ -196,6 +198,7 @@ public abstract class
SharedClusterCdcSidecarIntegrationTestBase extends SharedC
systemViews,
sidecarCdcStats,
() -> rangeManager,
+ cassandraBridgeProvider,
sidecarCdcClientProvider,
cdcOptions);
}
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 8981f7bd7..7b990e42c 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
@@ -27,8 +27,8 @@ import io.vertx.core.Handler;
import io.vertx.core.Promise;
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.bridge.CdcBridgeFactory;
import org.apache.cassandra.cdc.CdcLogMode;
import org.apache.cassandra.cdc.TypeCache;
import org.apache.cassandra.cdc.api.CdcOptions;
@@ -78,6 +78,7 @@ public class CdcPublisher implements
Handler<Message<Object>>, PeriodicTask
private final ICdcStats cdcStats;
private CdcManager cdcManager;
private final Provider<RangeManager> rangeManagerProvider;
+ private final Provider<CassandraBridge> cassandraBridgeProvider;
private final CachingSchemaStore schemaStore;
private KafkaPublisher<?> kafkaPublisher;
private final Provider<SidecarCdcClient> sidecarCdcClientProvider;
@@ -96,6 +97,7 @@ public class CdcPublisher implements
Handler<Message<Object>>, PeriodicTask
CdcSystemViewsDatabaseAccessor systemViews,
SidecarCdcStats sidecarCdcStats,
Provider<RangeManager> rangeManagerProvider,
+ Provider<CassandraBridge> cassandraBridgeProvider,
Provider<SidecarCdcClient> sidecarCdcClientProvider,
CachingSchemaStore schemaStore,
KafkaProducerFactory kafkaProducerFactory,
@@ -111,6 +113,7 @@ public class CdcPublisher implements
Handler<Message<Object>>, PeriodicTask
this.clusterConfigProvider = clusterConfigProvider;
this.cdcStats = cdcStats;
this.rangeManagerProvider = rangeManagerProvider;
+ this.cassandraBridgeProvider = cassandraBridgeProvider;
this.sidecarCdcClientProvider = sidecarCdcClientProvider;
this.schemaStore = schemaStore;
this.kafkaProducerFactory = kafkaProducerFactory;
@@ -134,10 +137,7 @@ public class CdcPublisher implements
Handler<Message<Object>>, PeriodicTask
public EventConsumer eventConsumer(CdcConfig conf)
{
- CassandraVersion version = CdcBridgeFactory.get(
- instanceMetadataFetcher.callOnFirstAvailableInstance(instance ->
- instance.delegate().nodeSettings()).releaseVersion()
- ).getVersion();
+ CassandraVersion version = cassandraBridgeProvider.get().getVersion();
if (conf.logOnly())
{
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 c39d55db1..bc7c60eb8 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
@@ -23,8 +23,8 @@ import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import com.google.inject.Provider;
import org.apache.cassandra.bridge.CassandraBridge;
-import org.apache.cassandra.bridge.CdcBridgeFactory;
import org.apache.cassandra.cdc.api.SchemaSupplier;
import org.apache.cassandra.sidecar.common.response.NodeSettings;
import org.apache.cassandra.sidecar.config.SidecarConfiguration;
@@ -51,15 +51,18 @@ import org.apache.cassandra.spark.utils.TableIdentifier;
public class CdcSchemaSupplier implements SchemaSupplier
{
private final InstanceMetadataFetcher instanceMetadataFetcher;
+ private final Provider<CassandraBridge> cassandraBridgeProvider;
private final CdcDatabaseAccessor cdcDatabaseAccessor;
private final SidecarConfiguration sidecarConfiguration;
private final ConcurrentHashMap<TableIdentifier, UUID> tableIdCache = new
ConcurrentHashMap<>();
public CdcSchemaSupplier(InstanceMetadataFetcher instanceMetadataFetcher,
+ Provider<CassandraBridge> cassandraBridgeProvider,
CdcDatabaseAccessor cdcDatabaseAccessor,
SidecarConfiguration sidecarConfiguration)
{
this.instanceMetadataFetcher = instanceMetadataFetcher;
+ this.cassandraBridgeProvider = cassandraBridgeProvider;
this.cdcDatabaseAccessor = cdcDatabaseAccessor;
this.sidecarConfiguration = sidecarConfiguration;
}
@@ -76,7 +79,7 @@ public class CdcSchemaSupplier implements SchemaSupplier
instance ->
instance.delegate().metadata().exportSchemaAsString());
NodeSettings nodeSettings =
instanceMetadataFetcher.callOnFirstAvailableInstance(
instance -> instance.delegate().nodeSettings());
- CassandraBridge cassandraBridge =
CdcBridgeFactory.get(nodeSettings.releaseVersion());
+ CassandraBridge cassandraBridge = cassandraBridgeProvider.get();
Partitioner partitioner = getPartitioner(nodeSettings);
boolean batchStatementsEnabled =
sidecarConfiguration.serviceConfiguration().cdcConfiguration().batchStatementsEnabled();
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/cdc/SidecarCdcOptions.java
b/server/src/main/java/org/apache/cassandra/sidecar/cdc/SidecarCdcOptions.java
index 0b9c8a213..76be71edb 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/cdc/SidecarCdcOptions.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/cdc/SidecarCdcOptions.java
@@ -21,6 +21,9 @@ package org.apache.cassandra.sidecar.cdc;
import java.time.Duration;
import java.util.Map;
+import com.google.inject.Provider;
+import com.google.inject.ProvisionException;
+import org.apache.cassandra.bridge.CassandraBridge;
import org.apache.cassandra.bridge.CassandraVersion;
import org.apache.cassandra.cdc.api.CdcOptions;
import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
@@ -38,11 +41,15 @@ public class SidecarCdcOptions implements CdcOptions
{
private final InstanceMetadataFetcher instanceMetadataFetcher;
private final CdcConfig conf;
+ private final Provider<CassandraBridge> cassandraBridgeProvider;
- public SidecarCdcOptions(InstanceMetadataFetcher instanceMetadataFetcher,
CdcConfig conf)
+ public SidecarCdcOptions(InstanceMetadataFetcher instanceMetadataFetcher,
+ CdcConfig conf,
+ Provider<CassandraBridge> cassandraBridgeProvider)
{
this.instanceMetadataFetcher = instanceMetadataFetcher;
this.conf = conf;
+ this.cassandraBridgeProvider = cassandraBridgeProvider;
}
@@ -62,9 +69,19 @@ public class SidecarCdcOptions implements CdcOptions
@Override
public CassandraVersion version()
{
- String releaseVersion =
instanceMetadataFetcher.callOnFirstAvailableInstance(
- instance ->
instance.delegate().nodeSettings().releaseVersion());
- return
CassandraVersion.fromVersion(releaseVersion).orElse(CassandraVersion.FOURZERO);
+ try
+ {
+ return cassandraBridgeProvider.get().getVersion();
+ }
+ catch (ProvisionException exception)
+ {
+ // The bridge binding rejects release versions it does not support
+ if (exception.getCause() instanceof IllegalArgumentException)
+ {
+ return CassandraVersion.FOURZERO;
+ }
+ throw exception;
+ }
}
/**
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 790b8978d..6c1b01e86 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
@@ -20,12 +20,11 @@ package org.apache.cassandra.sidecar.cdc;
import com.google.inject.Inject;
+import com.google.inject.Provider;
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.common.response.NodeSettings;
-import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
import org.apache.cassandra.spark.data.CqlTable;
/**
@@ -33,19 +32,18 @@ import org.apache.cassandra.spark.data.CqlTable;
*/
public class SidecarCqlToAvroSchemaConverter implements
CqlToAvroSchemaConverter
{
- private final InstanceMetadataFetcher instanceMetadataFetcher;
+ private final Provider<CassandraBridge> cassandraBridgeProvider;
@Inject
- public SidecarCqlToAvroSchemaConverter(InstanceMetadataFetcher
instanceMetadataFetcher)
+ public SidecarCqlToAvroSchemaConverter(Provider<CassandraBridge>
cassandraBridgeProvider)
{
- this.instanceMetadataFetcher = instanceMetadataFetcher;
+ this.cassandraBridgeProvider = cassandraBridgeProvider;
}
public CassandraBridge cassandraBridge()
{
- NodeSettings nodeSettings =
instanceMetadataFetcher.callOnFirstAvailableInstance(instance->
instance.delegate().nodeSettings());
- return CdcBridgeFactory.get(nodeSettings.releaseVersion());
+ return cassandraBridgeProvider.get();
}
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 797a897e9..09c33b691 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
@@ -36,6 +36,8 @@ import jakarta.ws.rs.DELETE;
import jakarta.ws.rs.GET;
import jakarta.ws.rs.PUT;
import jakarta.ws.rs.Path;
+import org.apache.cassandra.bridge.CassandraBridge;
+import org.apache.cassandra.bridge.CdcBridgeFactory;
import org.apache.cassandra.cdc.api.CdcOptions;
import org.apache.cassandra.cdc.api.SchemaSupplier;
import org.apache.cassandra.cdc.avro.CqlToAvroSchemaConverter;
@@ -157,10 +159,12 @@ public class CdcModule extends AbstractModule
CassandraClusterSchemaMonitor
cassandraClusterSchemaMonitorInstance(InstanceMetadataFetcher
instanceMetadataFetcher,
CdcDatabaseAccessor databaseAccessor,
DriverUnsupportedSchemaCache driverUnsupportedSchemaCache,
-
SidecarConfiguration configuration)
+
SidecarConfiguration configuration,
+
Provider<CassandraBridge> cassandraBridgeProvider)
{
return new CassandraClusterSchemaMonitor(instanceMetadataFetcher,
databaseAccessor,
- driverUnsupportedSchemaCache,
configuration);
+ driverUnsupportedSchemaCache,
configuration,
+ cassandraBridgeProvider);
}
@ProvidesIntoMap
@@ -173,10 +177,9 @@ public class CdcModule extends AbstractModule
@Provides
@Singleton
- CqlToAvroSchemaConverter cqlToAvroSchemaConverter(InstanceMetadataFetcher
instanceMetadataFetcher)
+ CqlToAvroSchemaConverter
cqlToAvroSchemaConverter(Provider<CassandraBridge> cassandraBridgeProvider)
{
-
- return new SidecarCqlToAvroSchemaConverter(instanceMetadataFetcher);
+ return new SidecarCqlToAvroSchemaConverter(cassandraBridgeProvider);
}
@Provides
@@ -187,6 +190,19 @@ public class CdcModule extends AbstractModule
return new CdcConfigImpl(vertx, cdcConfigAccessor);
}
+ /**
+ * The bridge for the Cassandra version the local instances run.
Deliberately unscoped, and injected as a
+ * {@link Provider}, so that it is resolved on each use: the instances may
be unreachable when Sidecar starts,
+ * and may be upgraded to another version while it runs.
+ */
+ @Provides
+ CassandraBridge cassandraBridge(InstanceMetadataFetcher
instanceMetadataFetcher)
+ {
+ String releaseVersion =
instanceMetadataFetcher.callOnFirstAvailableInstance(
+ instance ->
instance.delegate().nodeSettings().releaseVersion());
+ return CdcBridgeFactory.get(releaseVersion);
+ }
+
@Provides
@Singleton
SchemaStorePublisherFactory schemaStorePublisherFactory()
@@ -350,10 +366,11 @@ public class CdcModule extends AbstractModule
@Provides
@Singleton
public SchemaSupplier schemaSupplier(InstanceMetadataFetcher
instanceMetadataFetcher,
+ Provider<CassandraBridge>
cassandraBridgeProvider,
CdcDatabaseAccessor
cdcDatabaseAccessor,
SidecarConfiguration
sidecarConfiguration)
{
- return new CdcSchemaSupplier(instanceMetadataFetcher,
cdcDatabaseAccessor, sidecarConfiguration);
+ return new CdcSchemaSupplier(instanceMetadataFetcher,
cassandraBridgeProvider, cdcDatabaseAccessor, sidecarConfiguration);
}
@Provides
@@ -453,6 +470,7 @@ public class CdcModule extends AbstractModule
CdcSystemViewsDatabaseAccessor systemViews,
SidecarCdcStats sidecarCdcStats,
RangeManager rangeManager,
+ Provider<CassandraBridge>
cassandraBridgeProvider,
Provider<SidecarCdcClient>
sidecarCdcClientProvider,
CachingSchemaStore schemaStore,
KafkaProducerFactory kafkaProducerFactory,
@@ -469,6 +487,7 @@ public class CdcModule extends AbstractModule
systemViews,
sidecarCdcStats,
() -> rangeManager,
+ cassandraBridgeProvider,
sidecarCdcClientProvider,
schemaStore,
kafkaProducerFactory,
@@ -477,9 +496,11 @@ public class CdcModule extends AbstractModule
@Provides
@Singleton
- public CdcOptions cdcOptions(InstanceMetadataFetcher
instanceMetadataFetcher, CdcConfig conf)
+ public CdcOptions cdcOptions(InstanceMetadataFetcher
instanceMetadataFetcher,
+ CdcConfig conf,
+ Provider<CassandraBridge>
cassandraBridgeProvider)
{
- return new SidecarCdcOptions(instanceMetadataFetcher, conf);
+ return new SidecarCdcOptions(instanceMetadataFetcher, conf,
cassandraBridgeProvider);
}
@ProvidesIntoMap
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 054d08a7d..e3bd0a649 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
@@ -32,6 +32,7 @@ import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.inject.Provider;
import io.vertx.core.Promise;
import org.apache.cassandra.bridge.CassandraBridge;
import org.apache.cassandra.bridge.CdcBridge;
@@ -82,17 +83,20 @@ public class CassandraClusterSchemaMonitor implements
PeriodicTask
private final CopyOnWriteArrayList<Runnable> schemaChangeListeners = new
CopyOnWriteArrayList<>();
private final SidecarConfiguration sidecarConfiguration;
private final InstanceMetadataFetcher instanceFetcher;
+ private final Provider<CassandraBridge> cassandraBridgeProvider;
public CassandraClusterSchemaMonitor(InstanceMetadataFetcher
instanceFetcher,
CdcDatabaseAccessor databaseAccessor,
DriverUnsupportedSchemaCache
driverUnsupportedSchemaCache,
- SidecarConfiguration
sidecarConfiguration)
+ SidecarConfiguration
sidecarConfiguration,
+ Provider<CassandraBridge>
cassandraBridgeProvider)
{
this.instanceFetcher = instanceFetcher;
this.databaseAccessor = databaseAccessor;
this.driverUnsupportedSchemaCache = driverUnsupportedSchemaCache;
this.sidecarConfiguration = sidecarConfiguration;
+ this.cassandraBridgeProvider = cassandraBridgeProvider;
}
public void addSchemaChangeListener(Runnable listener)
@@ -103,7 +107,7 @@ public class CassandraClusterSchemaMonitor implements
PeriodicTask
public void refresh()
{
NodeSettings nodeSettings =
instanceFetcher.callOnFirstAvailableInstance(instance ->
instance.delegate().nodeSettings());
- CassandraBridge cassandraBridge =
CdcBridgeFactory.get(nodeSettings.releaseVersion());
+ CassandraBridge cassandraBridge = cassandraBridgeProvider.get();
CdcBridge cdcBridge = CdcBridgeFactory.getCdcBridge(cassandraBridge);
try
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 fd3b282b8..51c799d0d 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,6 +32,8 @@ 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;
@@ -101,6 +103,8 @@ public class CdcPublisherTests
@Mock
private Provider<RangeManager> rangeManager;
@Mock
+ private Provider<CassandraBridge> cassandraBridgeProvider;
+ @Mock
private SidecarCdcClient sidecarCdcClient;
@Mock
private CdcOptions cdcOptions;
@@ -135,6 +139,7 @@ public class CdcPublisherTests
systemViews,
sidecarCdcStats,
rangeManager,
+ cassandraBridgeProvider,
() -> sidecarCdcClient,
schemaStore,
kafkaProducerFactory,
@@ -188,7 +193,7 @@ public class CdcPublisherTests
new CdcPublisher(
vertx, executorPools, clusterConfigProvider, schemaSupplier,
instanceMetadataFetcher,
enabledConfig, databaseAccessor, cdcStats, systemViews,
freshSidecarCdcStats, rangeManager,
- () -> sidecarCdcClient, schemaStore, kafkaProducerFactory,
cdcOptions
+ cassandraBridgeProvider, () -> sidecarCdcClient, schemaStore,
kafkaProducerFactory, cdcOptions
);
verify(freshSidecarCdcStats).captureCdcEnabled();
@@ -205,7 +210,7 @@ public class CdcPublisherTests
new CdcPublisher(
vertx, executorPools, clusterConfigProvider, schemaSupplier,
instanceMetadataFetcher,
disabledConfig, databaseAccessor, cdcStats, systemViews,
freshSidecarCdcStats, rangeManager,
- () -> sidecarCdcClient, schemaStore, kafkaProducerFactory,
cdcOptions
+ cassandraBridgeProvider, () -> sidecarCdcClient, schemaStore,
kafkaProducerFactory, cdcOptions
);
verify(freshSidecarCdcStats).captureCdcDisabled();
@@ -272,6 +277,10 @@ 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(cassandraBridgeProvider.get()).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 8d41b019f..c67692a15 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
@@ -26,7 +26,9 @@ import java.util.concurrent.ExecutionException;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import com.google.inject.Provider;
import org.apache.cassandra.bridge.CassandraBridge;
+import org.apache.cassandra.bridge.CassandraVersion;
import org.apache.cassandra.bridge.CdcBridge;
import org.apache.cassandra.bridge.CdcBridgeFactory;
import org.apache.cassandra.sidecar.common.response.NodeSettings;
@@ -39,14 +41,12 @@ 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.Mockito.CALLS_REAL_METHODS;
import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
@@ -59,6 +59,8 @@ public class CdcSchemaSupplierTest
@Mock
private CdcDatabaseAccessor cdcDatabaseAccessor;
@Mock
+ private Provider<CassandraBridge> cassandraBridgeProvider;
+ @Mock
private CassandraBridge cassandraBridge;
@Mock
private CdcBridge cdcBridge;
@@ -95,8 +97,11 @@ public class CdcSchemaSupplierTest
when(serviceConfiguration.cdcConfiguration()).thenReturn(cdcConfiguration);
when(cdcConfiguration.batchStatementsEnabled()).thenReturn(true);
+
when(cassandraBridgeProvider.get()).thenReturn(CdcBridgeFactory.get(CassandraVersion.FOURONE));
+
cdcSchemaSupplier = new CdcSchemaSupplier(
instanceMetadataFetcher,
+ cassandraBridgeProvider,
cdcDatabaseAccessor,
sidecarConfiguration
);
@@ -259,7 +264,7 @@ public class CdcSchemaSupplierTest
}
@Test
- void testGetCdcEnabledTablesCallsCdcBridgeFactory()
+ void testGetCdcEnabledTablesUsesTheProvidedBridge()
{
String releaseVersion = "4.1.0";
NodeSettings nodeSettings = mockNodeSettings(releaseVersion,
"Murmur3Partitioner");
@@ -271,12 +276,9 @@ public class CdcSchemaSupplierTest
when(cdcDatabaseAccessor.partitioner()).thenReturn(Partitioner.Murmur3Partitioner);
when(cdcDatabaseAccessor.getTableId(any(TableIdentifier.class))).thenReturn(UUID.randomUUID());
- try (MockedStatic<CdcBridgeFactory> cdcBridgeFactory =
mockStatic(CdcBridgeFactory.class, CALLS_REAL_METHODS))
- {
- cdcSchemaSupplier.getTables();
+ cdcSchemaSupplier.getTables();
- cdcBridgeFactory.verify(() ->
CdcBridgeFactory.get(releaseVersion));
- }
+ verify(cassandraBridgeProvider).get();
}
private NodeSettings mockNodeSettings(String releaseVersion, String
partitioner)
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/cdc/SidecarCdcOptionsTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/cdc/SidecarCdcOptionsTest.java
index 29fd6d340..f721d5521 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/cdc/SidecarCdcOptionsTest.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/cdc/SidecarCdcOptionsTest.java
@@ -20,15 +20,25 @@ package org.apache.cassandra.sidecar.cdc;
import java.time.Duration;
import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Provider;
+import com.google.inject.Provides;
+import com.google.inject.ProvisionException;
+import org.apache.cassandra.bridge.CassandraBridge;
+import org.apache.cassandra.bridge.CassandraVersion;
+import org.apache.cassandra.bridge.CdcBridgeFactory;
import
org.apache.cassandra.sidecar.common.server.utils.MillisecondBoundConfiguration;
import
org.apache.cassandra.sidecar.common.server.utils.SecondBoundConfiguration;
import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -46,14 +56,49 @@ import static org.mockito.Mockito.when;
class SidecarCdcOptionsTest
{
private CdcConfig conf;
+ private InstanceMetadataFetcher instanceMetadataFetcher;
+ private Provider<CassandraBridge> cassandraBridgeProvider;
private SidecarCdcOptions options;
@BeforeEach
void setUp()
{
conf = mock(CdcConfig.class);
- InstanceMetadataFetcher instanceMetadataFetcher =
mock(InstanceMetadataFetcher.class);
- options = new SidecarCdcOptions(instanceMetadataFetcher, conf);
+ instanceMetadataFetcher = mock(InstanceMetadataFetcher.class);
+ cassandraBridgeProvider = mock(Provider.class);
+ options = new SidecarCdcOptions(instanceMetadataFetcher, conf,
cassandraBridgeProvider);
+ }
+
+ @Test
+ void versionIsTheVersionOfTheProvidedBridge()
+ {
+ CassandraBridge bridge = mock(CassandraBridge.class);
+ when(bridge.getVersion()).thenReturn(CassandraVersion.SIXZERO);
+ when(cassandraBridgeProvider.get()).thenReturn(bridge);
+
+ assertThat(options.version()).isEqualTo(CassandraVersion.SIXZERO);
+ }
+
+ @Test
+ void versionFallsBackToFourZeroForAnUnsupportedReleaseVersion()
+ {
+ // A Guice provider, as in Sidecar: it wraps what the binding throws
in a ProvisionException
+ Provider<CassandraBridge> provider = providerOf(() ->
CdcBridgeFactory.get("9.9.9"));
+
+ assertThat(new SidecarCdcOptions(instanceMetadataFetcher, conf,
provider).version()).isEqualTo(CassandraVersion.FOURZERO);
+ }
+
+ @Test
+ void versionPropagatesOtherBridgeFailures()
+ {
+ IllegalStateException failure = new IllegalStateException("No
Cassandra instance is available");
+ Provider<CassandraBridge> provider = providerOf(() -> {
+ throw failure;
+ });
+
+ assertThatThrownBy(() -> new
SidecarCdcOptions(instanceMetadataFetcher, conf, provider).version())
+ .isInstanceOf(ProvisionException.class)
+ .hasCause(failure);
}
@Test
@@ -90,4 +135,16 @@ class SidecarCdcOptionsTest
assertThat(options.maximumAge()).isEqualTo(Duration.ofSeconds(120));
}
+
+ private static Provider<CassandraBridge>
providerOf(Supplier<CassandraBridge> bridge)
+ {
+ return Guice.createInjector(new AbstractModule()
+ {
+ @Provides
+ CassandraBridge cassandraBridge()
+ {
+ return bridge.get();
+ }
+ }).getProvider(CassandraBridge.class);
+ }
}
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/modules/CdcModuleCassandraBridgeTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/modules/CdcModuleCassandraBridgeTest.java
new file mode 100644
index 000000000..75ae7f3a3
--- /dev/null
+++
b/server/src/test/java/org/apache/cassandra/sidecar/modules/CdcModuleCassandraBridgeTest.java
@@ -0,0 +1,127 @@
+/*
+ * 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.modules;
+
+import java.nio.ByteBuffer;
+import java.util.Map;
+import java.util.Set;
+
+import org.junit.jupiter.api.Test;
+
+import com.google.inject.Binding;
+import com.google.inject.Key;
+import com.google.inject.Scopes;
+import com.google.inject.spi.Elements;
+import org.apache.cassandra.bridge.CassandraBridge;
+import org.apache.cassandra.bridge.CassandraVersion;
+import org.apache.cassandra.bridge.CdcBridge;
+import org.apache.cassandra.bridge.CdcBridgeFactory;
+import org.apache.cassandra.cdc.TypeCache;
+import org.apache.cassandra.cdc.avro.CqlToAvroSchemaConverter;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+import org.apache.cassandra.spark.data.CqlField;
+import org.apache.cassandra.spark.data.CqlTable;
+import org.apache.cassandra.spark.data.ReplicationFactor;
+import org.apache.cassandra.spark.data.partitioner.Partitioner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for the {@link CassandraBridge} binding of {@link CdcModule}
+ */
+class CdcModuleCassandraBridgeTest
+{
+ private static final String KEYSPACE = "cdc_module_bridge_test_ks";
+ private static final ReplicationFactor REPLICATION_FACTOR =
+ new
ReplicationFactor(ReplicationFactor.ReplicationStrategy.NetworkTopologyStrategy,
Map.of("DC1", 3));
+
+ private final CdcModule module = new CdcModule();
+ private final InstanceMetadataFetcher instanceMetadataFetcher =
mock(InstanceMetadataFetcher.class);
+
+ @Test
+ void testBindingIsNotASingleton()
+ {
+ // A singleton would pin the bridge of the version first seen, and
fail injection while no instance is up
+ Binding<?> binding = Elements.getElements(module).stream()
+ .filter(element -> element instanceof
Binding)
+ .map(element -> (Binding<?>) element)
+ .filter(element ->
element.getKey().equals(Key.get(CassandraBridge.class)))
+ .findFirst()
+ .orElseThrow(AssertionError::new);
+
+ assertThat(Scopes.isSingleton(binding)).isFalse();
+ }
+
+ @Test
+ void testResolvesTheReleaseVersionOnEachCall()
+ {
+
when(instanceMetadataFetcher.callOnFirstAvailableInstance(any())).thenReturn("4.1.0",
"5.0.4");
+
+ CassandraBridge before =
module.cassandraBridge(instanceMetadataFetcher);
+ // e.g. after the instances are upgraded
+ CassandraBridge after =
module.cassandraBridge(instanceMetadataFetcher);
+
+
assertThat(before).isSameAs(CdcBridgeFactory.get(CassandraVersion.FOURONE));
+
assertThat(after).isSameAs(CdcBridgeFactory.get(CassandraVersion.FIVEZERO))
+ .isNotSameAs(before);
+ }
+
+ @Test
+ void testSharesClassLoaderWithCdcBridge()
+ {
+
when(instanceMetadataFetcher.callOnFirstAvailableInstance(any())).thenReturn("5.0.4");
+
+ CassandraBridge bridge =
module.cassandraBridge(instanceMetadataFetcher);
+ CdcBridge cdcBridge = CdcBridgeFactory.getCdcBridge(bridge);
+ CqlToAvroSchemaConverter avroSchemaConverter =
CdcBridgeFactory.getCqlToAvroSchemaConverter(bridge);
+
+ ClassLoader classLoader = bridge.getClass().getClassLoader();
+
assertThat(cdcBridge.getClass().getClassLoader()).isSameAs(classLoader);
+
assertThat(avroSchemaConverter.getClass().getClassLoader()).isSameAs(classLoader);
+ }
+
+ @Test
+ void testUdtCollectionFromParsedSchemaIsUsableByTypeCache()
+ {
+ // Mirrors the CDC flow: Sidecar parses the schema with its bridge,
the UDTs are registered
+ // in the CDC bridge's types (CassandraSchema#updateCdcSchema), and
TypeCache later resolves
+ // column types against them when deserializing/serializing mutations.
+
when(instanceMetadataFetcher.callOnFirstAvailableInstance(any())).thenReturn("5.0.4");
+ CassandraBridge bridge =
module.cassandraBridge(instanceMetadataFetcher);
+ String udt = "CREATE TYPE " + KEYSPACE + ".task_result (status text,
message text);";
+ String table = "CREATE TABLE " + KEYSPACE + ".task (id uuid PRIMARY
KEY, name text, " +
+ "results list<frozen<task_result>>) WITH cdc = true;";
+ CqlTable cqlTable = bridge.buildSchema(table, KEYSPACE,
REPLICATION_FACTOR,
+ Partitioner.Murmur3Partitioner,
Set.of(udt));
+ cqlTable.udts().forEach(u -> CdcBridgeFactory.get(bridge.getVersion())
+ .cassandraTypes()
+ .updateUDTs(KEYSPACE, u));
+
+ CqlField.CqlType type =
TypeCache.get(bridge.getVersion()).getType(KEYSPACE,
"list<frozen<task_result>>");
+
+ // an empty list is enough to build the element serializer, which
previously failed with a
+ // ClassCastException because the UDT was loaded by a different
classloader
+ ByteBuffer emptyList = ByteBuffer.allocate(4).putInt(0).flip();
+ assertThat(type.deserializeToJavaType(emptyList)).asList().isEmpty();
+ }
+}
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 f6e115bcf..e225efc6d 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
@@ -134,14 +134,14 @@ class CassandraClusterSchemaMonitorTest
when(mockDriverUnsupportedSchemaCache.getFullSchema()).thenReturn("");
when(mockDatabaseAccessor.partitioner()).thenReturn(Partitioner.Murmur3Partitioner);
when(mockDatabaseAccessor.getTableId(any(TableIdentifier.class))).thenReturn(UUID.randomUUID());
- mockCdcBridgeFactory.when(() ->
CdcBridgeFactory.get(anyString())).thenReturn(mockCassandraBridge);
mockCdcBridgeFactory.when(() ->
CdcBridgeFactory.getCdcBridge(any(CassandraBridge.class))).thenReturn(mockCdcBridge);
clusterSchema = new CassandraClusterSchemaMonitor(
mockInstanceFetcher,
mockDatabaseAccessor,
mockDriverUnsupportedSchemaCache,
- mockSidecarConfiguration
+ mockSidecarConfiguration,
+ () -> mockCassandraBridge
);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
