sarankk commented on code in PR #238:
URL:
https://github.com/apache/cassandra-analytics/pull/238#discussion_r4039597244
##########
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:
Nit: Shall we precompute these values in constructor and return here
##########
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();
+ }
+
+ /**
+ * @return the number of transient (witness) replicas across all
datacenters, {@code 0} when none are configured
+ */
+ public Integer getTransientReplicationFactor()
+ {
+ return
replication.values().stream().mapToInt(ReplicaCounts::transientReplicas).sum();
+ }
+
+ /**
+ * @return {@code true} if any datacenter is configured with transient
(witness) replicas
+ */
+ public boolean hasTransientReplicas()
+ {
+ return replication.values().stream().anyMatch(counts ->
counts.transientReplicas() > 0);
+ }
+
+ /**
+ * @param datacenter the datacenter to look up
+ * @return the number of transient (witness) replicas in {@code
datacenter}, {@code 0} when none are configured
+ */
+ public int getTransientReplicas(@NotNull String datacenter)
+ {
+ ReplicaCounts counts = replication.get(datacenter);
+ return counts == null ? 0 : counts.transientReplicas();
+ }
+
+ /**
+ * @param datacenter the datacenter to look up
+ * @return the number of replicas in {@code datacenter} holding the full
data set
+ * @throws IllegalArgumentException when {@code datacenter} has no
replication factor
+ */
+ public int getFullReplicas(@NotNull String datacenter)
+ {
+ return counts(datacenter).fullReplicas();
+ }
+
+ private ReplicaCounts counts(@NotNull String datacenter)
+ {
+ ReplicaCounts counts = replication.get(datacenter);
+ if (counts == null)
{
- if ("class".equals(entry.getKey()))
- {
- continue;
- }
- this.options.put(entry.getKey(), entry.getValue());
+ throw new IllegalArgumentException(String.format("Datacenter %s
not found in replication factor %s",
+ datacenter,
replication.keySet()));
}
+ return counts;
}
- public Integer getTotalReplicationFactor()
+ /**
+ * @return per-datacenter replica counts, unmodifiable
+ */
+ @NotNull
+ public Map<String, ReplicaCounts> getReplication()
{
- return options.values().stream()
- .mapToInt(Integer::intValue)
- .sum();
+ return replication;
}
+ /**
+ * @return per-datacenter total replica counts, including transient
(witness) replicas. Unmodifiable, and derived
+ * from {@link #getReplication()}.
+ */
@NotNull
public Map<String, Integer> getOptions()
{
- return options;
+ Map<String, Integer> options = new LinkedHashMap<>(replication.size());
Review Comment:
Similarly shall we precompute these and return here, instead of compute on
every call
##########
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();
+ }
+
+ /**
+ * @return the number of transient (witness) replicas across all
datacenters, {@code 0} when none are configured
+ */
+ public Integer getTransientReplicationFactor()
+ {
+ return
replication.values().stream().mapToInt(ReplicaCounts::transientReplicas).sum();
+ }
+
+ /**
+ * @return {@code true} if any datacenter is configured with transient
(witness) replicas
+ */
+ public boolean hasTransientReplicas()
+ {
+ return replication.values().stream().anyMatch(counts ->
counts.transientReplicas() > 0);
+ }
+
+ /**
+ * @param datacenter the datacenter to look up
+ * @return the number of transient (witness) replicas in {@code
datacenter}, {@code 0} when none are configured
+ */
+ public int getTransientReplicas(@NotNull String datacenter)
+ {
+ ReplicaCounts counts = replication.get(datacenter);
+ return counts == null ? 0 : counts.transientReplicas();
+ }
+
+ /**
+ * @param datacenter the datacenter to look up
+ * @return the number of replicas in {@code datacenter} holding the full
data set
+ * @throws IllegalArgumentException when {@code datacenter} has no
replication factor
+ */
+ public int getFullReplicas(@NotNull String datacenter)
+ {
+ return counts(datacenter).fullReplicas();
+ }
+
+ private ReplicaCounts counts(@NotNull String datacenter)
+ {
+ ReplicaCounts counts = replication.get(datacenter);
+ if (counts == null)
{
- if ("class".equals(entry.getKey()))
- {
- continue;
- }
- this.options.put(entry.getKey(), entry.getValue());
+ throw new IllegalArgumentException(String.format("Datacenter %s
not found in replication factor %s",
Review Comment:
We seem to throw this only for full replica count extraction and not for
transient replica, any reason why?
##########
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();
+ }
+
+ /**
+ * @return the number of transient (witness) replicas across all
datacenters, {@code 0} when none are configured
+ */
+ public Integer getTransientReplicationFactor()
+ {
+ return
replication.values().stream().mapToInt(ReplicaCounts::transientReplicas).sum();
+ }
+
+ /**
+ * @return {@code true} if any datacenter is configured with transient
(witness) replicas
+ */
+ public boolean hasTransientReplicas()
+ {
+ return replication.values().stream().anyMatch(counts ->
counts.transientReplicas() > 0);
+ }
+
+ /**
+ * @param datacenter the datacenter to look up
+ * @return the number of transient (witness) replicas in {@code
datacenter}, {@code 0} when none are configured
+ */
+ public int getTransientReplicas(@NotNull String datacenter)
+ {
+ ReplicaCounts counts = replication.get(datacenter);
+ return counts == null ? 0 : counts.transientReplicas();
+ }
+
+ /**
+ * @param datacenter the datacenter to look up
+ * @return the number of replicas in {@code datacenter} holding the full
data set
+ * @throws IllegalArgumentException when {@code datacenter} has no
replication factor
+ */
+ public int getFullReplicas(@NotNull String datacenter)
+ {
+ return counts(datacenter).fullReplicas();
+ }
+
+ private ReplicaCounts counts(@NotNull String datacenter)
+ {
+ ReplicaCounts counts = replication.get(datacenter);
+ if (counts == null)
{
- if ("class".equals(entry.getKey()))
- {
- continue;
- }
- this.options.put(entry.getKey(), entry.getValue());
+ throw new IllegalArgumentException(String.format("Datacenter %s
not found in replication factor %s",
+ datacenter,
replication.keySet()));
}
+ return counts;
}
- public Integer getTotalReplicationFactor()
+ /**
+ * @return per-datacenter replica counts, unmodifiable
+ */
+ @NotNull
+ public Map<String, ReplicaCounts> getReplication()
{
- return options.values().stream()
- .mapToInt(Integer::intValue)
- .sum();
+ return replication;
}
+ /**
+ * @return per-datacenter total replica counts, including transient
(witness) replicas. Unmodifiable, and derived
+ * from {@link #getReplication()}.
+ */
@NotNull
public Map<String, Integer> getOptions()
{
- return options;
+ Map<String, Integer> options = new LinkedHashMap<>(replication.size());
+ replication.forEach((datacenter, counts) -> options.put(datacenter,
counts.allReplicas()));
+ return Collections.unmodifiableMap(options);
+ }
+
+ /**
+ * @return per-datacenter transient (witness) replica counts,
unmodifiable. Datacenters without transient
+ * replicas are absent.
+ */
+ @NotNull
+ public Map<String, Integer> getTransientOptions()
+ {
+ Map<String, Integer> transientOptions = new LinkedHashMap<>();
Review Comment:
Same as above
--
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]
