From 323b84be6fec23cebda06e455cfb6b239fe2f634 Mon Sep 17 00:00:00 2001 From: Bianca Stanciu Date: Mon, 13 Jul 2026 16:24:35 +0300 Subject: [PATCH] [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() {