mansikhara commented on code in PR #238:
URL:
https://github.com/apache/cassandra-analytics/pull/238#discussion_r4041080934
##########
cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/ReplicationFactor.java:
##########
@@ -106,62 +110,209 @@ public static ReplicationFactor simpleStrategy(int rf)
@NotNull
private final ReplicationStrategy replicationStrategy;
+ /**
+ * Per-datacenter replica counts. Holding the total and the transient
count together, rather than in two
+ * parallel maps, means the two cannot drift apart.
+ */
@NotNull
- private final Map<String, Integer> options;
+ private final Map<String, ReplicaCounts> replication;
+ /**
+ * Parses a raw replication map. A value that cannot be parsed, or that
Cassandra itself would reject, raises
+ * {@link IllegalArgumentException} naming the offending datacenter: a
partial replication factor is not usable
+ * and reporting it here avoids a misleading failure later, such as "DC
not found in replication factor".
+ *
+ * @param options the raw replication map, including the {@code class}
entry
+ * @throws IllegalArgumentException when any replication value cannot be
parsed
+ */
public ReplicationFactor(@NotNull Map<String, String> options)
{
- this.replicationStrategy =
ReplicationFactor.ReplicationStrategy.getEnum(options.get("class"));
- this.options = new LinkedHashMap<>(options.size());
- for (Map.Entry<String, String> entry : options.entrySet())
+ this(ReplicationStrategy.getEnum(options.get("class")),
parse(options), options);
+ }
+
+ public ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy,
@NotNull Map<String, Integer> options)
+ {
+ this(replicationStrategy, options, Collections.emptyMap());
+ }
+
+ public ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy,
+ @NotNull Map<String, Integer> options,
+ @NotNull Map<String, Integer> transientOptions)
+ {
+ this(replicationStrategy, merge(options, transientOptions), options);
+ }
+
+ /**
+ * Canonical constructor. Every other constructor resolves its input to
per-datacenter counts and delegates here.
+ *
+ * @param replicationStrategy the replication strategy
+ * @param replication per-datacenter replica counts, already
validated individually
+ * @param raw the caller's original input, used only in
the error message
+ */
+ private ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy,
+ @NotNull Map<String, ReplicaCounts> replication,
+ @NotNull Object raw)
+ {
+ // A strategy other than LocalStrategy with no datacenter entries is
not usable
+ if (replicationStrategy != ReplicationStrategy.LocalStrategy &&
replication.isEmpty())
{
- if ("class".equals(entry.getKey()))
+ throw new IllegalArgumentException("Could not find replication
info in schema map: " + raw);
+ }
+ this.replicationStrategy = replicationStrategy;
+ this.replication = Collections.unmodifiableMap(new
LinkedHashMap<>(replication));
+ }
+
+ /**
+ * Resolves raw string values, e.g. {@code "3"} or {@code "3/1"}, to
per-datacenter counts.
+ */
+ private static Map<String, ReplicaCounts> parse(@NotNull Map<String,
String> options)
+ {
+ Map<String, ReplicaCounts> parsed = new
LinkedHashMap<>(options.size());
+ options.forEach((datacenter, value) -> {
+ if ("class".equals(datacenter))
{
- continue;
+ return;
}
-
try
{
- this.options.put(entry.getKey(),
Integer.parseInt(entry.getValue()));
+ parsed.put(datacenter, ReplicaCounts.parse(value));
}
- catch (NumberFormatException exception)
+ catch (IllegalArgumentException exception)
{
- LOGGER.warn("Could not parse replication option: {} = {}",
entry.getKey(), entry.getValue());
+ throw new IllegalArgumentException(String.format("Could not
parse replication option: %s = %s",
+ datacenter,
value), exception);
}
- }
+ });
+ return parsed;
}
- public ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy,
@NotNull Map<String, Integer> options)
+ /**
+ * Resolves separate total and transient maps to per-datacenter counts,
rejecting a transient entry for a
+ * datacenter that has no replication factor.
+ */
+ private static Map<String, ReplicaCounts> merge(@NotNull Map<String,
Integer> options,
+ @NotNull Map<String,
Integer> transientOptions)
{
- this.replicationStrategy = replicationStrategy;
- this.options = new LinkedHashMap<>(options.size());
+ Map<String, ReplicaCounts> merged = new
LinkedHashMap<>(options.size());
+ options.forEach((datacenter, allReplicas) -> {
+ if (!"class".equals(datacenter))
+ {
+ merged.put(datacenter,
+ ReplicaCounts.of(datacenter, allReplicas,
transientOptions.getOrDefault(datacenter, 0)));
+ }
+ });
+ transientOptions.forEach((datacenter, transientReplicas) -> {
+ if (!"class".equals(datacenter) && transientReplicas != null &&
transientReplicas != 0
+ && !merged.containsKey(datacenter))
+ {
+ throw new IllegalArgumentException(String.format(
+ "Transient replicas specified for %s but it has no replication
factor", datacenter));
+ }
+ });
+ return merged;
+ }
- if (!replicationStrategy.equals(ReplicationStrategy.LocalStrategy) &&
options.isEmpty())
- {
- throw new RuntimeException(String.format("Could not find
replication info in schema map: %s.", options));
- }
+ /**
+ * @return the total number of replicas across all datacenters, including
transient (witness) replicas.
+ * Semantics are unchanged from before transient replica support
was added.
+ */
+ public Integer getTotalReplicationFactor()
+ {
+ return
replication.values().stream().mapToInt(ReplicaCounts::allReplicas).sum();
+ }
- for (Map.Entry<String, Integer> entry : options.entrySet())
+ /**
+ * @return the number of replicas across all datacenters that hold the
full data set, i.e. the total
+ * replication factor minus transient (witness) replicas
+ */
+ public Integer getFullReplicationFactor()
+ {
+ return
replication.values().stream().mapToInt(ReplicaCounts::fullReplicas).sum();
Review Comment:
Done!
--
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]
