mansikhara commented on code in PR #238:
URL:
https://github.com/apache/cassandra-analytics/pull/238#discussion_r4032254935
##########
cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/CqlUtilsTest.java:
##########
@@ -142,6 +142,83 @@ public void testExtractReplicationFactor(CassandraBridge
bridge)
assertThat(systemSchemaRf.getOptions()).isEqualTo(ImmutableMap.of());
}
+ @Test
+ public void testExtractReplicationFactorWithWitnessReplicas()
+ {
+ // The <replicas>/<transient> form, as Cassandra's
WitnessAlwaysReadsFullReplicaTest creates it on the
+ // cep-45-mutation-tracking branch. replication_type is deliberately
omitted: Sidecar builds its schema
+ // response from the driver's exportAsString(), which does not emit
that property, and it is irrelevant to
+ // replication factor parsing in any case. extractReplicationType is
covered separately.
Review Comment:
Done!
##########
cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/ReplicationFactor.java:
##########
@@ -106,62 +110,204 @@ 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());
+ Map<String, ReplicaCounts> parsed = new
LinkedHashMap<>(options.size());
for (Map.Entry<String, String> entry : options.entrySet())
{
if ("class".equals(entry.getKey()))
{
continue;
}
-
try
{
- this.options.put(entry.getKey(),
Integer.parseInt(entry.getValue()));
+ parsed.put(entry.getKey(),
ReplicaCounts.parse(entry.getValue()));
}
- 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",
+
entry.getKey(), entry.getValue()), exception);
}
}
+ validateNotEmpty(this.replicationStrategy, parsed, options);
+ this.replication = Collections.unmodifiableMap(parsed);
}
public ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy,
@NotNull Map<String, Integer> options)
{
- this.replicationStrategy = replicationStrategy;
- this.options = new LinkedHashMap<>(options.size());
-
- if (!replicationStrategy.equals(ReplicationStrategy.LocalStrategy) &&
options.isEmpty())
- {
- throw new RuntimeException(String.format("Could not find
replication info in schema map: %s.", options));
- }
+ this(replicationStrategy, options, Collections.emptyMap());
+ }
+ public ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy,
+ @NotNull Map<String, Integer> options,
+ @NotNull Map<String, Integer> transientOptions)
+ {
+ this.replicationStrategy = replicationStrategy;
+ Map<String, ReplicaCounts> merged = new
LinkedHashMap<>(options.size());
for (Map.Entry<String, Integer> entry : options.entrySet())
{
if ("class".equals(entry.getKey()))
{
continue;
}
- this.options.put(entry.getKey(), entry.getValue());
+ int transientReplicas =
transientOptions.getOrDefault(entry.getKey(), 0);
+ merged.put(entry.getKey(), ReplicaCounts.of(entry.getKey(),
entry.getValue(), transientReplicas));
+ }
+ for (String datacenter : transientOptions.keySet())
+ {
+ if (!"class".equals(datacenter) && !merged.containsKey(datacenter)
+ && transientOptions.get(datacenter) != null &&
transientOptions.get(datacenter) != 0)
+ {
+ throw new IllegalArgumentException(String.format(
+ "Transient replicas specified for %s but it has no replication
factor", datacenter));
+ }
+ }
+ validateNotEmpty(replicationStrategy, merged, options);
+ this.replication = Collections.unmodifiableMap(merged);
+ }
+
+ /**
+ * Canonical constructor. {@code validate} exists only for {@link
#readResolve()}, which must not reject an
+ * instance that an earlier version was willing to create.
+ */
+ private ReplicationFactor(@NotNull ReplicationStrategy replicationStrategy,
+ @NotNull Map<String, ReplicaCounts> replication,
+ boolean validate)
+ {
+ this.replicationStrategy = replicationStrategy;
+ if (validate)
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]
