From 323b84be6fec23cebda06e455cfb6b239fe2f634 Mon Sep 17 00:00:00 2001 From: Bianca Stanciu Date: Mon, 13 Jul 2026 16:24:35 +0300 Subject: [PATCH 1/2] [CASSANALYTICS-181] - Support Sidecar behind a load balancer for time-skew validation in coordinated writes --- CHANGES.txt | 1 + .../spark/bulkwriter/BulkSparkConf.java | 7 +++ .../bulkwriter/CassandraClusterInfo.java | 31 ++++++---- .../spark/bulkwriter/WriterOptions.java | 9 +++ .../CassandraClusterInfoGroup.java | 26 ++++++++- .../CoordinatedCassandraClusterInfo.java | 57 +++++++++++++++++++ .../bulkwriter/CassandraClusterInfoTest.java | 39 +++++++++++++ 7 files changed, 158 insertions(+), 12 deletions(-) create mode 100644 cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/coordinated/CoordinatedCassandraClusterInfo.java diff --git a/CHANGES.txt b/CHANGES.txt index 49ed57ae5..d8dd54aa9 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,6 @@ 0.5.0 ----- + * Support Sidecar behind a load balancer for time-skew validation in coordinated writes (CASSANALYTICS-181) * SSTable-version-based bridge determination (CASSANALYTICS-24) * Upgrade sidecar version to 0.4.0 (CASSANALYTICS-176) * Exclude IP address from RingInstance equality so node replacement does not fail bulk write jobs (CASSANALYTICS-175) diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java index 6e64f145a..21e4b57d5 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java @@ -146,6 +146,7 @@ public class BulkSparkConf implements Serializable public final double importCoordinatorTimeoutMultiplier; public boolean quoteIdentifiers; public final boolean skipSecondaryIndexCheck; + public final boolean sidecarBehindLoadBalancer; protected final String keystorePassword; protected final String keystorePath; protected final String keystoreBase64Encoded; @@ -229,6 +230,7 @@ public BulkSparkConf(SparkConf conf, Map options, @Nullable Logg this.timestamp = MapUtils.getOrDefault(options, WriterOptions.TIMESTAMP.name(), null); this.quoteIdentifiers = MapUtils.getBoolean(options, WriterOptions.QUOTE_IDENTIFIERS.name(), false, "quote identifiers"); this.skipSecondaryIndexCheck = MapUtils.getBoolean(options, WriterOptions.SKIP_SECONDARY_INDEX_CHECK.name(), false, "skip secondary index check"); + this.sidecarBehindLoadBalancer = MapUtils.getBoolean(options, WriterOptions.SIDECAR_BEHIND_LOAD_BALANCER.name(), false, "sidecar behind load balancer"); int storageClientConcurrency = MapUtils.getInt(options, WriterOptions.STORAGE_CLIENT_CONCURRENCY.name(), DEFAULT_STORAGE_CLIENT_CONCURRENCY, "storage client concurrency"); long storageClientKeepAliveSeconds = MapUtils.getLong(options, WriterOptions.STORAGE_CLIENT_THREAD_KEEP_ALIVE_SECONDS.name(), @@ -276,6 +278,11 @@ public BulkSparkConf(SparkConf conf, Map options, @Nullable Logg this.configuredJobId = MapUtils.getOrDefault(options, WriterOptions.JOB_ID.name(), null); this.coordinatedWriteConfJson = MapUtils.getOrDefault(options, WriterOptions.COORDINATED_WRITE_CONFIG.name(), null); this.coordinatedWriteConf = buildCoordinatedWriteConf(dataTransportInfo.getTransport(), logger); + if (this.sidecarBehindLoadBalancer && this.coordinatedWriteConf == null && logger != null) + { + logger.warn("{} is set but {} is not configured; the flag is only honored for coordinated writes and will be ignored on the single-cluster path.", + WriterOptions.SIDECAR_BEHIND_LOAD_BALANCER, WriterOptions.COORDINATED_WRITE_CONFIG); + } this.digestAlgorithmSupplier = digestAlgorithmSupplierFromOptions(dataTransport, options); validateEnvironment(); } diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java index 1b1a40179..4ff53dc20 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java @@ -246,16 +246,7 @@ void validateTimeSkewWithLocalNow(Range range, Instant localNow) thr TimeSkewResponse timeSkew; try { - TokenRangeMapping topology = getTokenRangeMapping(true); - List instances = topology.getSubRanges(range) - .asMapOfRanges() - .values() - .stream() - .flatMap(Collection::stream) - .distinct() // remove duplications - .map(replica -> new SidecarInstanceImpl(replica.nodeName(), getCassandraContext().sidecarPort())) - .collect(Collectors.toList()); - timeSkew = getCassandraContext().getSidecarClient().timeSkew(instances).get(); + timeSkew = fetchTimeSkew(range).get(); } catch (InterruptedException | ExecutionException exception) { @@ -270,6 +261,26 @@ void validateTimeSkewWithLocalNow(Range range, Instant localNow) thr } } + /** + * Fetches time-skew information from Sidecar. The default implementation queries the replicas + * that own {@code range}. Subclasses may override to target a different endpoint set — for + * example, coordinated writes route through shared contact points when replica FQDNs are not + * reachable from Spark executors. + */ + protected CompletableFuture fetchTimeSkew(Range range) + { + TokenRangeMapping topology = getTokenRangeMapping(true); + List instances = topology.getSubRanges(range) + .asMapOfRanges() + .values() + .stream() + .flatMap(Collection::stream) + .distinct() // remove duplications + .map(replica -> new SidecarInstanceImpl(replica.nodeName(), getCassandraContext().sidecarPort())) + .collect(Collectors.toList()); + return getCassandraContext().getSidecarClient().timeSkew(instances); + } + @Override public synchronized void refreshClusterInfo() { diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/WriterOptions.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/WriterOptions.java index 5440016ac..e907c4f0a 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/WriterOptions.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/WriterOptions.java @@ -154,4 +154,13 @@ public enum WriterOptions implements WriterOption * */ STORAGE_CREDENTIAL_TYPE, + /** + * Option declaring that the Sidecar cluster is fronted by a load balancer, so replica FQDNs + * from the token map are not directly routable from Spark executors. When {@code true}, + * requests that would otherwise fan out to per-replica Sidecar addresses are routed through + * the configured contact points instead. Defaults to {@code false}. + *

+ * Today this affects time-skew validation under coordinated write. + */ + SIDECAR_BEHIND_LOAD_BALANCER, } diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/coordinated/CassandraClusterInfoGroup.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/coordinated/CassandraClusterInfoGroup.java index 5032844cd..ecbc19810 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/coordinated/CassandraClusterInfoGroup.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/coordinated/CassandraClusterInfoGroup.java @@ -95,7 +95,29 @@ public class CassandraClusterInfoGroup implements ClusterInfo, MultiClusterSuppo */ public static CassandraClusterInfoGroup fromBulkSparkConf(BulkSparkConf conf) { - return fromBulkSparkConf(conf, clusterId -> new CassandraClusterInfo(conf, clusterId)); + return fromBulkSparkConf(conf, clusterId -> createClusterInfo(conf, clusterId)); + } + + /** + * Selects the {@link CassandraClusterInfo} concrete type based on the + * {@link BulkSparkConf#sidecarBehindLoadBalancer} flag. Kept centralized so the + * driver-side factory and the executor-side broadcast reconstruction stay in lockstep. + */ + private static CassandraClusterInfo createClusterInfo(BulkSparkConf conf, String clusterId) + { + if (conf.sidecarBehindLoadBalancer) + { + LOGGER.info("Using CoordinatedCassandraClusterInfo for load-balanced Sidecar. clusterId={}", clusterId); + return new CoordinatedCassandraClusterInfo(conf, clusterId); + } + return new CassandraClusterInfo(conf, clusterId); + } + + private static CassandraClusterInfo createClusterInfo(BroadcastableClusterInfo bci) + { + return bci.getConf().sidecarBehindLoadBalancer + ? new CoordinatedCassandraClusterInfo(bci) + : new CassandraClusterInfo(bci); } /** @@ -170,7 +192,7 @@ private CassandraClusterInfoGroup(BroadcastableClusterInfoGroup broadcastable) // Build list of ClusterInfo from broadcastable data List clusterInfosList = new ArrayList<>(); broadcastable.forEach((clusterId, broadcastableInfo) -> { - clusterInfosList.add(new CassandraClusterInfo((BroadcastableClusterInfo) broadcastableInfo)); + clusterInfosList.add(createClusterInfo((BroadcastableClusterInfo) broadcastableInfo)); }); this.clusterInfos = Collections.unmodifiableList(clusterInfosList); diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/coordinated/CoordinatedCassandraClusterInfo.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/coordinated/CoordinatedCassandraClusterInfo.java new file mode 100644 index 000000000..996580f64 --- /dev/null +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/coordinated/CoordinatedCassandraClusterInfo.java @@ -0,0 +1,57 @@ +/* + * 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.spark.bulkwriter.cloudstorage.coordinated; + +import java.math.BigInteger; +import java.util.concurrent.CompletableFuture; + +import com.google.common.collect.Range; + +import o.a.c.sidecar.client.shaded.common.response.TimeSkewResponse; +import org.apache.cassandra.spark.bulkwriter.BroadcastableClusterInfo; +import org.apache.cassandra.spark.bulkwriter.BulkSparkConf; +import org.apache.cassandra.spark.bulkwriter.CassandraClusterInfo; + +/** + * Variant of {@link CassandraClusterInfo} used when Sidecars are fronted by a load balancer. + * Replica FQDNs from the token map are not routable from Spark executors in that topology, + * so time-skew validation must query the configured contact points (load balancer endpoints) + * rather than per-range replicas. Selected when + * {@link org.apache.cassandra.spark.bulkwriter.WriterOptions#SIDECAR_BEHIND_LOAD_BALANCER} + * is set. + */ +public class CoordinatedCassandraClusterInfo extends CassandraClusterInfo +{ + public CoordinatedCassandraClusterInfo(BulkSparkConf conf, String clusterId) + { + super(conf, clusterId); + } + + public CoordinatedCassandraClusterInfo(BroadcastableClusterInfo broadcastable) + { + super(broadcastable); + } + + @Override + protected CompletableFuture fetchTimeSkew(Range range) + { + return getCassandraContext().getSidecarClient().timeSkew(); + } +} diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java index 5ed4c5221..aa4d4b5bd 100644 --- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java @@ -27,7 +27,9 @@ import com.google.common.collect.ImmutableMap; import org.junit.jupiter.api.Test; +import o.a.c.sidecar.client.shaded.client.SidecarClient; import o.a.c.sidecar.client.shaded.common.response.TimeSkewResponse; +import org.apache.cassandra.spark.bulkwriter.cloudstorage.coordinated.CoordinatedCassandraClusterInfo; import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping; import org.apache.cassandra.spark.exception.TimeSkewTooLargeException; @@ -35,8 +37,11 @@ import static org.assertj.core.api.Assertions.assertThatNoException; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.Mockito.RETURNS_DEEP_STUBS; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; public class CassandraClusterInfoTest @@ -53,6 +58,40 @@ void testTimeSkewAcceptable() .isThrownBy(() -> ci.validateTimeSkewWithLocalNow(range(10, 20), localNow)); } + @Test + void testCoordinatedClusterInfoUsesSidecarClientContactPointsForTimeSkew() + { + Instant localNow = Instant.now(); + int allowanceMinutes = 10; + + SidecarClient sidecarClient = mock(SidecarClient.class); + CassandraContext ctx = mock(CassandraContext.class); + when(ctx.getSidecarClient()).thenReturn(sidecarClient); + TimeSkewResponse tsr = new TimeSkewResponse(localNow.toEpochMilli(), allowanceMinutes); + when(sidecarClient.timeSkew()).thenReturn(CompletableFuture.completedFuture(tsr)); + + CassandraClusterInfo ci = new CoordinatedCassandraClusterInfo((BulkSparkConf) null, null) + { + @Override + protected CassandraContext buildCassandraContext() + { + return ctx; + } + + @Override + public TokenRangeMapping getTokenRangeMapping(boolean cached) + { + return TokenRangeMappingUtils.buildTokenRangeMapping(0, ImmutableMap.of("dc1", 3), 5); + } + }; + + assertThatNoException() + .describedAs("Load-balanced cluster info must use timeSkew() so load balancer contact points stay reachable") + .isThrownBy(() -> ci.validateTimeSkewWithLocalNow(range(10, 20), localNow)); + verify(sidecarClient).timeSkew(); + verify(sidecarClient, never()).timeSkew(anyList()); + } + @Test void testTimeSkewTooLarge() { From f4bc953dff133222800ecf2ddee9095300e2f10f Mon Sep 17 00:00:00 2001 From: Bianca Stanciu Date: Mon, 24 Aug 2026 17:51:19 +0300 Subject: [PATCH 2/2] [CASSANALYTICS-181] - Support Sidecar behind a load balancer for time-skew validation in single-cluster and coordinated writes --- CHANGES.txt | 2 +- .../bulkwriter/BroadcastableClusterInfo.java | 2 +- .../spark/bulkwriter/BulkSparkConf.java | 5 -- .../CassandraBulkWriterContext.java | 2 +- .../bulkwriter/CassandraClusterInfo.java | 51 +++++++++++++++++++ ... => LoadBalancedCassandraClusterInfo.java} | 21 ++++---- .../spark/bulkwriter/WriterOptions.java | 2 +- .../CassandraClusterInfoGroup.java | 26 +--------- .../bulkwriter/CassandraClusterInfoTest.java | 44 ++++++++++++++-- 9 files changed, 110 insertions(+), 45 deletions(-) rename cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/{cloudstorage/coordinated/CoordinatedCassandraClusterInfo.java => LoadBalancedCassandraClusterInfo.java} (64%) diff --git a/CHANGES.txt b/CHANGES.txt index e6dc7cf4e..226c89cb1 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,6 +1,6 @@ 0.5.0 ----- - * Support Sidecar behind a load balancer for time-skew validation in coordinated writes (CASSANALYTICS-181) + * Support Sidecar behind a load balancer for time-skew validation in single-cluster and coordinated writes (CASSANALYTICS-181) * CDC reader stats silently dropped in SidecarCdcBuilder (CASSANALYTICS-191) * Add CapturePublishedSchema metric to SidecarCdcStats (CASSANALYTICS-189) * Expand list of architecture that supports unaligned access in FastByteOperations (CASSANALYTICS-188) diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BroadcastableClusterInfo.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BroadcastableClusterInfo.java index b180c9a09..e40ced51a 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BroadcastableClusterInfo.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BroadcastableClusterInfo.java @@ -100,6 +100,6 @@ public String clusterId() @Override public ClusterInfo reconstruct() { - return new CassandraClusterInfo(this); + return CassandraClusterInfo.create(this); } } diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java index 21e4b57d5..ccfb457b0 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java @@ -278,11 +278,6 @@ public BulkSparkConf(SparkConf conf, Map options, @Nullable Logg this.configuredJobId = MapUtils.getOrDefault(options, WriterOptions.JOB_ID.name(), null); this.coordinatedWriteConfJson = MapUtils.getOrDefault(options, WriterOptions.COORDINATED_WRITE_CONFIG.name(), null); this.coordinatedWriteConf = buildCoordinatedWriteConf(dataTransportInfo.getTransport(), logger); - if (this.sidecarBehindLoadBalancer && this.coordinatedWriteConf == null && logger != null) - { - logger.warn("{} is set but {} is not configured; the flag is only honored for coordinated writes and will be ignored on the single-cluster path.", - WriterOptions.SIDECAR_BEHIND_LOAD_BALANCER, WriterOptions.COORDINATED_WRITE_CONFIG); - } this.digestAlgorithmSupplier = digestAlgorithmSupplierFromOptions(dataTransport, options); validateEnvironment(); } diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraBulkWriterContext.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraBulkWriterContext.java index 92721c5a4..9d740c23f 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraBulkWriterContext.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraBulkWriterContext.java @@ -59,7 +59,7 @@ protected CassandraBulkWriterContext(@NotNull BulkWriterConfig config) @Override protected ClusterInfo buildClusterInfo() { - return new CassandraClusterInfo(bulkSparkConf()); + return CassandraClusterInfo.create(bulkSparkConf()); } @Override diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java index 93da6f78e..64e9e5453 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java @@ -138,6 +138,57 @@ public CassandraClusterInfo(BroadcastableClusterInfo broadcastable) this.allNodeSettingFutures = null; } + /** + * Creates a {@link CassandraClusterInfo} for a single cluster, selecting the concrete type based on the + * {@link BulkSparkConf#sidecarBehindLoadBalancer} flag. + * + * @param conf bulk write conf + * @return {@link LoadBalancedCassandraClusterInfo} when Sidecar is behind a load balancer, otherwise a plain + * {@link CassandraClusterInfo} + */ + public static CassandraClusterInfo create(BulkSparkConf conf) + { + return create(conf, null); + } + + /** + * Creates a {@link CassandraClusterInfo}, selecting the concrete type based on the + * {@link BulkSparkConf#sidecarBehindLoadBalancer} flag. Kept centralized so the driver-side factory and the + * executor-side broadcast reconstruction stay in lockstep, for both single-cluster and coordinated writes. + * + * @param conf bulk write conf + * @param clusterId cluster identifier, or {@code null} for a single (non-coordinated) cluster + * @return {@link LoadBalancedCassandraClusterInfo} when Sidecar is behind a load balancer, otherwise a plain + * {@link CassandraClusterInfo} + */ + public static CassandraClusterInfo create(BulkSparkConf conf, String clusterId) + { + if (conf.sidecarBehindLoadBalancer) + { + LOGGER.info("Using LoadBalancedCassandraClusterInfo for load-balanced Sidecar. clusterId={}", clusterId); + return new LoadBalancedCassandraClusterInfo(conf, clusterId); + } + return new CassandraClusterInfo(conf, clusterId); + } + + /** + * Reconstructs a {@link CassandraClusterInfo} on an executor from broadcast, selecting the concrete type based on + * the {@link BulkSparkConf#sidecarBehindLoadBalancer} flag so it matches the driver-side selection in + * {@link #create(BulkSparkConf, String)}. + * + * @param broadcastable the broadcastable cluster info from broadcast + * @return {@link LoadBalancedCassandraClusterInfo} when Sidecar is behind a load balancer, otherwise a plain + * {@link CassandraClusterInfo} + */ + public static CassandraClusterInfo create(BroadcastableClusterInfo broadcastable) + { + if (broadcastable.getConf().sidecarBehindLoadBalancer) + { + return new LoadBalancedCassandraClusterInfo(broadcastable); + } + return new CassandraClusterInfo(broadcastable); + } + @Override public void checkBulkWriterIsEnabledOrThrow() { diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/coordinated/CoordinatedCassandraClusterInfo.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/LoadBalancedCassandraClusterInfo.java similarity index 64% rename from cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/coordinated/CoordinatedCassandraClusterInfo.java rename to cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/LoadBalancedCassandraClusterInfo.java index 996580f64..7a8b176af 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/coordinated/CoordinatedCassandraClusterInfo.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/LoadBalancedCassandraClusterInfo.java @@ -17,7 +17,7 @@ * under the License. */ -package org.apache.cassandra.spark.bulkwriter.cloudstorage.coordinated; +package org.apache.cassandra.spark.bulkwriter; import java.math.BigInteger; import java.util.concurrent.CompletableFuture; @@ -25,26 +25,27 @@ import com.google.common.collect.Range; import o.a.c.sidecar.client.shaded.common.response.TimeSkewResponse; -import org.apache.cassandra.spark.bulkwriter.BroadcastableClusterInfo; -import org.apache.cassandra.spark.bulkwriter.BulkSparkConf; -import org.apache.cassandra.spark.bulkwriter.CassandraClusterInfo; /** * Variant of {@link CassandraClusterInfo} used when Sidecars are fronted by a load balancer. * Replica FQDNs from the token map are not routable from Spark executors in that topology, - * so time-skew validation must query the configured contact points (load balancer endpoints) - * rather than per-range replicas. Selected when + * so requests that would otherwise fan out to per-replica Sidecar addresses are routed through + * the configured contact points (load balancer endpoints) instead. + *

+ * This applies to both single-cluster and coordinated writes. It is selected by + * {@link CassandraClusterInfo#create(BulkSparkConf, String)} and + * {@link CassandraClusterInfo#create(BroadcastableClusterInfo)} when * {@link org.apache.cassandra.spark.bulkwriter.WriterOptions#SIDECAR_BEHIND_LOAD_BALANCER} * is set. */ -public class CoordinatedCassandraClusterInfo extends CassandraClusterInfo +public class LoadBalancedCassandraClusterInfo extends CassandraClusterInfo { - public CoordinatedCassandraClusterInfo(BulkSparkConf conf, String clusterId) + public LoadBalancedCassandraClusterInfo(BulkSparkConf conf, String clusterId) { super(conf, clusterId); } - public CoordinatedCassandraClusterInfo(BroadcastableClusterInfo broadcastable) + public LoadBalancedCassandraClusterInfo(BroadcastableClusterInfo broadcastable) { super(broadcastable); } @@ -52,6 +53,8 @@ public CoordinatedCassandraClusterInfo(BroadcastableClusterInfo broadcastable) @Override protected CompletableFuture fetchTimeSkew(Range range) { + // range is irrelevant; the load balancer contact points are queried directly rather + // than the per-range replicas, whose FQDNs are not routable from Spark executors. return getCassandraContext().getSidecarClient().timeSkew(); } } diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/WriterOptions.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/WriterOptions.java index e907c4f0a..142b7b1e1 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/WriterOptions.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/WriterOptions.java @@ -160,7 +160,7 @@ public enum WriterOptions implements WriterOption * requests that would otherwise fan out to per-replica Sidecar addresses are routed through * the configured contact points instead. Defaults to {@code false}. *

- * Today this affects time-skew validation under coordinated write. + * Today this affects time-skew validation, and is honored for both single-cluster and coordinated writes. */ SIDECAR_BEHIND_LOAD_BALANCER, } diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/coordinated/CassandraClusterInfoGroup.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/coordinated/CassandraClusterInfoGroup.java index ecbc19810..c80a56034 100644 --- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/coordinated/CassandraClusterInfoGroup.java +++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/coordinated/CassandraClusterInfoGroup.java @@ -95,29 +95,7 @@ public class CassandraClusterInfoGroup implements ClusterInfo, MultiClusterSuppo */ public static CassandraClusterInfoGroup fromBulkSparkConf(BulkSparkConf conf) { - return fromBulkSparkConf(conf, clusterId -> createClusterInfo(conf, clusterId)); - } - - /** - * Selects the {@link CassandraClusterInfo} concrete type based on the - * {@link BulkSparkConf#sidecarBehindLoadBalancer} flag. Kept centralized so the - * driver-side factory and the executor-side broadcast reconstruction stay in lockstep. - */ - private static CassandraClusterInfo createClusterInfo(BulkSparkConf conf, String clusterId) - { - if (conf.sidecarBehindLoadBalancer) - { - LOGGER.info("Using CoordinatedCassandraClusterInfo for load-balanced Sidecar. clusterId={}", clusterId); - return new CoordinatedCassandraClusterInfo(conf, clusterId); - } - return new CassandraClusterInfo(conf, clusterId); - } - - private static CassandraClusterInfo createClusterInfo(BroadcastableClusterInfo bci) - { - return bci.getConf().sidecarBehindLoadBalancer - ? new CoordinatedCassandraClusterInfo(bci) - : new CassandraClusterInfo(bci); + return fromBulkSparkConf(conf, clusterId -> CassandraClusterInfo.create(conf, clusterId)); } /** @@ -192,7 +170,7 @@ private CassandraClusterInfoGroup(BroadcastableClusterInfoGroup broadcastable) // Build list of ClusterInfo from broadcastable data List clusterInfosList = new ArrayList<>(); broadcastable.forEach((clusterId, broadcastableInfo) -> { - clusterInfosList.add(createClusterInfo((BroadcastableClusterInfo) broadcastableInfo)); + clusterInfosList.add(CassandraClusterInfo.create((BroadcastableClusterInfo) broadcastableInfo)); }); this.clusterInfos = Collections.unmodifiableList(clusterInfosList); diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java index 96c4e48d4..9452944da 100644 --- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java @@ -24,11 +24,13 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.stream.Stream; import com.google.common.collect.ImmutableMap; +import com.google.common.collect.Maps; import com.google.common.util.concurrent.Uninterruptibles; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; @@ -39,11 +41,12 @@ import o.a.c.sidecar.client.shaded.client.SidecarClient; import o.a.c.sidecar.client.shaded.common.response.NodeSettings; import o.a.c.sidecar.client.shaded.common.response.TimeSkewResponse; -import org.apache.cassandra.spark.bulkwriter.cloudstorage.coordinated.CoordinatedCassandraClusterInfo; import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping; import org.apache.cassandra.spark.exception.TimeSkewTooLargeException; +import org.apache.spark.SparkConf; import static org.apache.cassandra.spark.TestUtils.range; +import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatNoException; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; @@ -69,7 +72,7 @@ void testTimeSkewAcceptable() } @Test - void testCoordinatedClusterInfoUsesSidecarClientContactPointsForTimeSkew() + void testLoadBalancedClusterInfoUsesSidecarClientContactPointsForTimeSkew() { Instant localNow = Instant.now(); int allowanceMinutes = 10; @@ -80,7 +83,7 @@ void testCoordinatedClusterInfoUsesSidecarClientContactPointsForTimeSkew() TimeSkewResponse tsr = new TimeSkewResponse(localNow.toEpochMilli(), allowanceMinutes); when(sidecarClient.timeSkew()).thenReturn(CompletableFuture.completedFuture(tsr)); - CassandraClusterInfo ci = new CoordinatedCassandraClusterInfo((BulkSparkConf) null, null) + CassandraClusterInfo ci = new LoadBalancedCassandraClusterInfo((BulkSparkConf) null, null) { @Override protected CassandraContext buildCassandraContext() @@ -102,6 +105,41 @@ public TokenRangeMapping getTokenRangeMapping(boolean cached) verify(sidecarClient, never()).timeSkew(anyList()); } + @Test + void testCreateSelectsLoadBalancedClusterInfoWhenSidecarBehindLoadBalancer() + { + // Single-cluster path: the SIDECAR_BEHIND_LOAD_BALANCER flag must be honored, not just for coordinated writes + try (CassandraClusterInfo ci = CassandraClusterInfo.create(bulkSparkConf(true))) + { + assertThat(ci) + .describedAs("Sidecar behind a load balancer must select the load-balanced variant on the single-cluster path") + .isExactlyInstanceOf(LoadBalancedCassandraClusterInfo.class); + } + } + + @Test + void testCreateSelectsPlainClusterInfoByDefault() + { + try (CassandraClusterInfo ci = CassandraClusterInfo.create(bulkSparkConf(false))) + { + assertThat(ci) + .describedAs("Without the load balancer flag, the plain per-replica variant must be selected") + .isExactlyInstanceOf(CassandraClusterInfo.class); + } + } + + private static BulkSparkConf bulkSparkConf(boolean sidecarBehindLoadBalancer) + { + // No keystore options: this keeps SSL disabled so create() can build a real (non-TLS) Sidecar client + // offline. We only assert on the concrete type create() selects, never issuing a request. + Map options = Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER); + options.put(WriterOptions.SIDECAR_CONTACT_POINTS.name(), "127.0.0.1"); + options.put(WriterOptions.KEYSPACE.name(), "ks"); + options.put(WriterOptions.TABLE.name(), "table"); + options.put(WriterOptions.SIDECAR_BEHIND_LOAD_BALANCER.name(), String.valueOf(sidecarBehindLoadBalancer)); + return new BulkSparkConf(new SparkConf(), options); + } + @Test void testTimeSkewTooLarge() {