skoppu22 commented on code in PR #223:
URL:
https://github.com/apache/cassandra-analytics/pull/223#discussion_r3720163253
##########
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)
Review Comment:
Same as above
##########
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();
Review Comment:
The same problem exists for Single cluster behind a load balancer. Hence
instead of using 'Coordinated' word, good to develop this as a generic
framework for handling cluster/clusters behind a load balancer, then we can
invoke it from anywhere needed
##########
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);
Review Comment:
This name CoordinatedCassandraClusterInfo is misleading. For the reader,
this naming sounds like Coordinated work is happening only when
sidecarBehindLoadBalancer is true, and else case below doing non-coordinated
work. Need to pick appropriate name instead of CoordinatedCassandraClusterInfo
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java:
##########
@@ -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.",
Review Comment:
What about single cluster behind a load balancer? Can you create a Jira for
supporting this case and add Jira here as a TODO ?
##########
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();
Review Comment:
Also, range param is unused, hence mention a comment something like 'range
is irrelevant; contact points are queried directly'
##########
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
Review Comment:
As I mentioned above, this naming is incorrect, this class is for Cluster
behind a LB, nothing specific to Corodinated
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
