yifan-c commented on code in PR #169:
URL:
https://github.com/apache/cassandra-analytics/pull/169#discussion_r3534369239
##########
cassandra-analytics-sidecar-client/src/main/java/org/apache/cassandra/clients/Sidecar.java:
##########
@@ -149,6 +153,79 @@ public static List<CompletableFuture<NodeSettings>>
allNodeSettings(SidecarClien
.collect(Collectors.toList());
}
+ /**
+ * Retrieve gossip info from all nodes on the cluster
+ * @param client Sidecar client
+ * @param instances all Sidecar instances
+ * @return completable futures with GossipInfoResponse
+ */
+ public static List<CompletableFuture<GossipInfoResponse>>
gossipInfoFromAllNodes(SidecarClient client,
+
Set<SidecarInstance> instances)
+ {
+ return instances.stream()
+ .map(instance -> client
+ .gossipInfo(instance)
+ .exceptionally(throwable -> {
+ LOGGER.warn(String.format("Failed to retrieve
gossipinfo from instance=%s",
+ instance), throwable);
+ return null;
+ }))
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Retrieves SSTable versions from all nodes in the cluster via gossip
info.
+ * This method fetches gossip information from all Sidecar instances and
extracts
+ * the SSTable versions running on each node.
+ *
+ * @param client Sidecar client
+ * @param instances all Sidecar instances in the cluster
+ * @param maxRetryDelayMillis maximum delay in milliseconds between retries
+ * @param maxRetries maximum number of retry attempts
+ * @return a set of SSTable versions across all nodes in the cluster
+ * @throws RuntimeException if unable to retrieve gossip info from any
nodes
+ */
+ public static Set<String> getSSTableVersionsFromCluster(SidecarClient
client,
+
Set<SidecarInstance> instances,
+ long
maxRetryDelayMillis,
+ int maxRetries)
+ {
+ LOGGER.debug("Retrieving SSTable versions from cluster via gossip...");
+
+ List<CompletableFuture<GossipInfoResponse>> gossipInfoFutures =
+ gossipInfoFromAllNodes(client, instances);
+
+ // Calculate total timeout
+ final long totalTimeout = maxRetryDelayMillis * maxRetries *
gossipInfoFutures.size();
Review Comment:
The total timeout indicates the sequential execution, but the requests are
indeed executed in parallel. I think you want to remove
`gossipInfoFutures.size()` in the calculation.
##########
cassandra-analytics-sidecar-client/src/main/java/org/apache/cassandra/clients/Sidecar.java:
##########
@@ -149,6 +153,79 @@ public static List<CompletableFuture<NodeSettings>>
allNodeSettings(SidecarClien
.collect(Collectors.toList());
}
+ /**
+ * Retrieve gossip info from all nodes on the cluster
+ * @param client Sidecar client
+ * @param instances all Sidecar instances
+ * @return completable futures with GossipInfoResponse
+ */
+ public static List<CompletableFuture<GossipInfoResponse>>
gossipInfoFromAllNodes(SidecarClient client,
+
Set<SidecarInstance> instances)
+ {
+ return instances.stream()
+ .map(instance -> client
+ .gossipInfo(instance)
+ .exceptionally(throwable -> {
+ LOGGER.warn(String.format("Failed to retrieve
gossipinfo from instance=%s",
+ instance), throwable);
+ return null;
+ }))
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Retrieves SSTable versions from all nodes in the cluster via gossip
info.
+ * This method fetches gossip information from all Sidecar instances and
extracts
+ * the SSTable versions running on each node.
+ *
+ * @param client Sidecar client
+ * @param instances all Sidecar instances in the cluster
+ * @param maxRetryDelayMillis maximum delay in milliseconds between retries
+ * @param maxRetries maximum number of retry attempts
+ * @return a set of SSTable versions across all nodes in the cluster
+ * @throws RuntimeException if unable to retrieve gossip info from any
nodes
+ */
+ public static Set<String> getSSTableVersionsFromCluster(SidecarClient
client,
+
Set<SidecarInstance> instances,
+ long
maxRetryDelayMillis,
+ int maxRetries)
Review Comment:
This and many other changes in the patch do not follow the codestyle of the
project.
You can run `./gradlew idea` to re-generate the project, and run `crtl +
option + i` (or in the menu `Code -> Auto-Indent Lines`) in each changed files.
##########
cassandra-analytics-common/src/main/java/org/apache/cassandra/bridge/CassandraVersion.java:
##########
@@ -78,13 +109,69 @@ public String jarBaseName()
return jarBaseName;
}
- private static final String sstableFormat;
+ /**
+ * Get the set of SSTable formats supported by this Cassandra version.
+ *
+ * @return Set of supported SSTable format strings
+ */
+ public Set<String> sstableFormats()
+ {
+ return sstableFormats;
+ }
+
+ /**
+ * Get the list of native SSTable version strings for this Cassandra
version.
+ *
+ * @return List of native SSTable version strings
+ */
+ public List<String> getNativeSStableVersions()
+ {
+ return nativeSStableVersions;
+ }
+
+ /**
+ * Get the set of SSTable version strings that this Cassandra version can
read.
+ * This includes the native versions of every Cassandra version from
{@link #lowestCompatibleVersionNumber}
+ * (inclusive) up to this version (inclusive). For example, Cassandra 5.0
can read 4.0, 4.1 and 5.0
+ * SSTables (lowestCompatible=40) but NOT 3.x.
+ *
+ * @return Set of full SSTable version strings that can be read
+ */
+ public Set<String> getSupportedSStableVersionsForRead()
Review Comment:
`SStable` -> `SSTable`
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/bridge/SSTableVersionAnalyzer.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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.bridge;
+
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import org.apache.cassandra.spark.bulkwriter.BulkSparkConf;
+
+/**
+ * Determines which Cassandra bridge to load from the SSTable versions present
on a cluster.
+ */
+public final class SSTableVersionAnalyzer
+{
+ private SSTableVersionAnalyzer()
+ {
+ // Utility class
+ }
+
+ /**
+ * Determines the bridge version for bulk write from the SSTable versions
present on the cluster.
+ * Picks the lowest Cassandra version present so the produced SSTables can
be imported by every node.
+ *
+ * @param sstableVersionsOnCluster set of SSTable versions found on
cluster nodes
+ * @param requestedFormat user's requested format, e.g. "big" or
"bti"
+ * @return the CassandraVersion bridge to load for write
+ * @throws IllegalStateException if the versions are empty/null or
contain an unknown version
+ * @throws UnsupportedOperationException if the versions are mutually
incompatible, or the chosen
+ * bridge does not support the
requested format
+ */
+ public static CassandraVersion determineBridgeVersionForWrite(Set<String>
sstableVersionsOnCluster, String requestedFormat)
+ {
+ requireSSTableVersionsPresent(sstableVersionsOnCluster);
+
+ // Every observed version must be mutually compatible (readable by the
highest version present).
+ ensureMutuallyCompatibleVersions(sstableVersionsOnCluster,
highestCompatibleVersion(sstableVersionsOnCluster));
+
+ CassandraVersion bridgeVersion =
cassandraVersionsFromSSTableVersions(sstableVersionsOnCluster)
+
.min(Comparator.comparingInt(CassandraVersion::versionNumber))
+ .orElseThrow(() -> new
IllegalStateException("Unable to determine the lowest SSTable version"));
Review Comment:
In this method, `sstableVersionsOnCluster` is converted to
`CassandraVersion` twice.
##########
cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/MockBulkWriterContext.java:
##########
@@ -378,9 +379,17 @@ public void checkBulkWriterIsEnabledOrThrow()
}
@Override
- public String getLowestCassandraVersion()
+ public CassandraVersion getBridgeVersion()
{
- return cassandraVersion;
+ return
CassandraVersion.fromVersion(cassandraVersion).orElse(CassandraVersion.FIVEZERO);
+ }
+
+ public Set<String> getSSTableVersionsOnCluster()
+ {
+ // Return SSTable versions based on Cassandra version for testing
+ CassandraVersion version =
CassandraVersion.fromVersion(cassandraVersion)
+ .orElse(CassandraVersion.FIVEZERO);
+ return new HashSet<>(version.getNativeSStableVersions());
}
Review Comment:
not used
##########
cassandra-analytics-common/src/main/java/org/apache/cassandra/bridge/CassandraVersion.java:
##########
@@ -78,13 +109,69 @@ public String jarBaseName()
return jarBaseName;
}
- private static final String sstableFormat;
+ /**
+ * Get the set of SSTable formats supported by this Cassandra version.
+ *
+ * @return Set of supported SSTable format strings
+ */
+ public Set<String> sstableFormats()
+ {
+ return sstableFormats;
+ }
+
+ /**
+ * Get the list of native SSTable version strings for this Cassandra
version.
+ *
+ * @return List of native SSTable version strings
+ */
+ public List<String> getNativeSStableVersions()
Review Comment:
`SStable` -> `SSTable`
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java:
##########
@@ -844,6 +934,67 @@ private static String readNullable(ObjectInputStream in)
throws IOException
return null;
}
+ /**
+ * Validates that all SSTables being read have versions that were observed
in gossip info.
+ * This catches cases where SSTables have unexpected versions that weren't
seen during driver initialization.
+ * This validation runs on executors.
+ *
+ * @param sstables list of SSTables to validate
+ * @throws UnsupportedOperationException if any SSTable has a version not
in expected sstable versions
+ */
+ @VisibleForTesting
+ void validateSStableVersions(List<SSTable> sstables)
Review Comment:
`SStable` -> `SSTable`
--
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]
