[CASSANALYTICS-181] - Support Sidecar behind a load balancer for time…

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)

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<String, String> 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<String, String> 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();
     }

cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java

@@ -246,16 +246,7 @@ void validateTimeSkewWithLocalNow(Range<BigInteger> range, Instant localNow) thr
         TimeSkewResponse timeSkew;
         try
         {
-            TokenRangeMapping<RingInstance> topology = getTokenRangeMapping(true);
-            List<SidecarInstance> 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<BigInteger> 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<TimeSkewResponse> fetchTimeSkew(Range<BigInteger> range)
+    {
+        TokenRangeMapping<RingInstance> topology = getTokenRangeMapping(true);
+        List<SidecarInstance> 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()
     {

cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/WriterOptions.java

@@ -154,4 +154,13 @@ public enum WriterOptions implements WriterOption
      * </ul>
      */
     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}.
+     * <p>
+     * Today this affects time-skew validation under coordinated write.
+     */
+    SIDECAR_BEHIND_LOAD_BALANCER,
 }

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<ClusterInfo> clusterInfosList = new ArrayList<>();
         broadcastable.forEach((clusterId, broadcastableInfo) -> {
-            clusterInfosList.add(new CassandraClusterInfo((BroadcastableClusterInfo) broadcastableInfo));
+            clusterInfosList.add(createClusterInfo((BroadcastableClusterInfo) broadcastableInfo));
         });
         this.clusterInfos = Collections.unmodifiableList(clusterInfosList);
 

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<TimeSkewResponse> fetchTimeSkew(Range<BigInteger> range)
+    {
+        return getCassandraContext().getSidecarClient().timeSkew();
+    }
+}

cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java

@@ -27,16 +27,21 @@
 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;
 
 import static org.apache.cassandra.spark.TestUtils.range;
 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<RingInstance> 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()
     {