skoppu22 commented on code in PR #169:
URL:
https://github.com/apache/cassandra-analytics/pull/169#discussion_r3475947031
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/cloudstorage/coordinated/CassandraClusterInfoGroup.java:
##########
@@ -398,9 +369,62 @@ private RuntimeException toRuntimeException(ClusterInfo
clusterInfo, Throwable c
return new RuntimeException("Failed to perform action on cluster: " +
clusterInfo.clusterId(), cause);
}
- @VisibleForTesting
- Map<String, ClusterInfo> clusterInfoByIdUnsafe()
+ /**
+ * Sets the bridge version on all contained CassandraClusterInfo instances.
+ *
+ * @param bridgeVersion the determined Cassandra bridge version
+ */
+ public void setBridgeVersion(CassandraVersion bridgeVersion)
{
- return clusterInfoById;
+ for (ClusterInfo ci : clusterInfos)
+ {
+ ((CassandraClusterInfo) ci).setBridgeVersion(bridgeVersion);
+ }
+ }
+
+ /**
+ * Retrieves the lowest Cassandra version from all contained clusters.
+ *
+ * @return lowest Cassandra version string across all clusters
+ */
+ public String getLowestCassandraVersion()
+ {
+ Map<String, String> clusterVersions = new HashMap<>();
+ for (ClusterInfo ci : clusterInfos)
+ {
+ CassandraClusterInfo cci = (CassandraClusterInfo) ci;
+ clusterVersions.put(ci.clusterId(),
cci.getLowestCassandraVersion());
+ }
+
+ // Find the lowest version across all clusters
+ List<CassandraVersionFeatures> versions = clusterVersions.values()
+ .stream()
+
.map(CassandraVersionFeatures::cassandraVersionFeaturesFromCassandraVersion)
+ .sorted()
+
.collect(Collectors.toList());
+
+ CassandraVersionFeatures first = versions.get(0);
+ CassandraVersionFeatures last = versions.get(versions.size() - 1);
+ Preconditions.checkState(first.getMajorVersion() ==
last.getMajorVersion(),
+ "Cluster versions are not compatible.
lowest=%s and highest=%s",
+ first.getRawVersionString(),
last.getRawVersionString());
+
+ return first.getRawVersionString();
+ }
+
+ /**
+ * Retrieves aggregated SSTable versions from all contained clusters.
+ *
+ * @return set of SSTable versions present across all clusters
+ */
+ public Set<String> getSSTableVersionsOnCluster()
+ {
+ Set<String> aggregatedSSTableVersions = new HashSet<>();
+ for (ClusterInfo ci : clusterInfos)
+ {
+ CassandraClusterInfo cci = (CassandraClusterInfo) ci;
+
aggregatedSSTableVersions.addAll(cci.getSSTableVersionsOnCluster());
+ }
+ return aggregatedSSTableVersions;
}
Review Comment:
getSSTableVersionsOnCluster is called only on driver. Executors get
determined bridge version. So there is no cached value to return.
--
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]
