jyothsnakonisa commented on code in PR #238:
URL:
https://github.com/apache/cassandra-analytics/pull/238#discussion_r4077978962
##########
cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/ReplicationFactor.java:
##########
@@ -188,13 +359,173 @@ public boolean equals(Object other)
ReplicationFactor that = (ReplicationFactor) other;
return this.replicationStrategy == that.replicationStrategy
- && java.util.Objects.equals(this.options, that.options);
+ && Objects.equals(this.replication, that.replication);
}
@Override
public int hashCode()
{
- return Objects.hash(replicationStrategy, options);
+ return Objects.hash(replicationStrategy, replication);
+ }
+
+ /**
+ * {@link #serialVersionUID} is pinned, so an instance serialized before
the per-datacenter counts were combined
+ * deserializes with a {@code null} {@code replication}. Only reachable
under driver/executor version skew, which
+ * is not a supported configuration, so this reports the mismatch rather
than silently losing the counts.
+ *
+ * @return this instance
+ * @throws InvalidObjectException when deserialized from an incompatible
older form
+ */
+ private Object readResolve() throws InvalidObjectException
+ {
+ if (replication == null)
+ {
+ throw new InvalidObjectException(
+ "ReplicationFactor was serialized by an incompatible version:
per-datacenter replica counts are absent. "
+ + "Driver and executors must run the same version.");
+ }
+ // Rebuild through the canonical constructor so the derived fields,
which are transient, are populated
+ return new ReplicationFactor(replicationStrategy, replication,
replication);
+ }
+
+ /**
+ * Replica counts for a single datacenter. Cassandra accepts either {@code
<replicas>} or
+ * {@code <replicas>/<transient>}; transient replication predates mutation
tracking, and witness replicas
+ * (CEP-46) reuse the same form. See {@code
org.apache.cassandra.locator.ReplicationFactor} in Cassandra.
+ */
+ public static final class ReplicaCounts implements Serializable
+ {
+ private static final long serialVersionUID = 2026091500000000001L;
+ private static final String TRANSIENT_SEPARATOR = "/";
+
+ private final int allReplicas;
+ private final int transientReplicas;
+
+ private ReplicaCounts(int allReplicas, int transientReplicas)
+ {
+ this.allReplicas = allReplicas;
+ this.transientReplicas = transientReplicas;
+ }
+
+ /**
+ * @param datacenter datacenter name, used only in error
messages, may be {@code null}
+ * @param allReplicas total replicas
+ * @param transientReplicas transient (witness) replicas
+ * @return validated counts
+ * @throws IllegalArgumentException when the counts are inconsistent
+ */
+ public static ReplicaCounts of(String datacenter, int allReplicas, int
transientReplicas)
+ {
+ validate(datacenter, allReplicas, transientReplicas);
+ return new ReplicaCounts(allReplicas, transientReplicas);
+ }
+
+ /**
+ * @return total replicas, including transient (witness) replicas
+ */
+ public int allReplicas()
+ {
+ return allReplicas;
+ }
+
+ /**
+ * @return replicas holding the full data set
+ */
+ public int fullReplicas()
+ {
+ return allReplicas - transientReplicas;
+ }
+
+ /**
+ * @return transient (witness) replicas, {@code 0} when none are
configured
+ */
+ public int transientReplicas()
+ {
+ return transientReplicas;
+ }
+
+ /**
+ * @param value the raw replication value, e.g. {@code "3"} or {@code
"3/1"}
+ * @return the parsed replica counts
+ * @throws NumberFormatException when either component is not an
integer
+ * @throws IllegalArgumentException when the value is malformed or the
counts are inconsistent
+ */
+ static ReplicaCounts parse(@NotNull String value)
+ {
+ String trimmed = value.trim();
+ int separator = trimmed.indexOf(TRANSIENT_SEPARATOR);
+ if (separator < 0)
+ {
+ return of(null, Integer.parseInt(trimmed), 0);
+ }
+
+ if (trimmed.indexOf(TRANSIENT_SEPARATOR, separator + 1) >= 0)
+ {
+ throw new IllegalArgumentException(String.format(
+ "Replication factor format is <replicas> or
<replicas>/<transient>, found '%s'", value));
+ }
+
+ return of(null,
+ Integer.parseInt(trimmed.substring(0, separator).trim()),
+ Integer.parseInt(trimmed.substring(separator +
1).trim()));
+ }
+
+ /**
+ * Mirrors the constraints Cassandra enforces in {@code
ReplicationFactor.validate}: transient replicas must
+ * be non-negative and strictly fewer than the total, so at least one
full replica always exists.
+ *
+ * @param datacenter datacenter name for the error message, may
be {@code null}
+ * @param allReplicas total replicas
+ * @param transientReplicas transient (witness) replicas
+ */
+ static void validate(String datacenter, int allReplicas, int
transientReplicas)
Review Comment:
ReplicaCounts class doesn't need `datacenter` at all and the datacenter
argument fit here in this method. I see that you are passing this for logging
can u try moving logging out of this class
##########
cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/ReplicationFactor.java:
##########
@@ -188,13 +359,173 @@ public boolean equals(Object other)
ReplicationFactor that = (ReplicationFactor) other;
return this.replicationStrategy == that.replicationStrategy
- && java.util.Objects.equals(this.options, that.options);
+ && Objects.equals(this.replication, that.replication);
}
@Override
public int hashCode()
{
- return Objects.hash(replicationStrategy, options);
+ return Objects.hash(replicationStrategy, replication);
+ }
+
+ /**
+ * {@link #serialVersionUID} is pinned, so an instance serialized before
the per-datacenter counts were combined
+ * deserializes with a {@code null} {@code replication}. Only reachable
under driver/executor version skew, which
+ * is not a supported configuration, so this reports the mismatch rather
than silently losing the counts.
+ *
+ * @return this instance
+ * @throws InvalidObjectException when deserialized from an incompatible
older form
+ */
+ private Object readResolve() throws InvalidObjectException
+ {
+ if (replication == null)
+ {
+ throw new InvalidObjectException(
+ "ReplicationFactor was serialized by an incompatible version:
per-datacenter replica counts are absent. "
+ + "Driver and executors must run the same version.");
+ }
+ // Rebuild through the canonical constructor so the derived fields,
which are transient, are populated
+ return new ReplicationFactor(replicationStrategy, replication,
replication);
+ }
+
+ /**
+ * Replica counts for a single datacenter. Cassandra accepts either {@code
<replicas>} or
+ * {@code <replicas>/<transient>}; transient replication predates mutation
tracking, and witness replicas
+ * (CEP-46) reuse the same form. See {@code
org.apache.cassandra.locator.ReplicationFactor} in Cassandra.
+ */
+ public static final class ReplicaCounts implements Serializable
+ {
+ private static final long serialVersionUID = 2026091500000000001L;
+ private static final String TRANSIENT_SEPARATOR = "/";
+
+ private final int allReplicas;
+ private final int transientReplicas;
+
+ private ReplicaCounts(int allReplicas, int transientReplicas)
+ {
+ this.allReplicas = allReplicas;
+ this.transientReplicas = transientReplicas;
+ }
+
+ /**
+ * @param datacenter datacenter name, used only in error
messages, may be {@code null}
+ * @param allReplicas total replicas
+ * @param transientReplicas transient (witness) replicas
+ * @return validated counts
+ * @throws IllegalArgumentException when the counts are inconsistent
+ */
+ public static ReplicaCounts of(String datacenter, int allReplicas, int
transientReplicas)
Review Comment:
ReplicaCounts class doesn't need datacenter at all and the datacenter
argument fit here in this method. I see that you are passing this for logging
can u try moving logging out of this class
##########
cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/ReplicationFactor.java:
##########
@@ -203,25 +534,29 @@ public static class Serializer extends
com.esotericsoftware.kryo.Serializer<Repl
public void write(Kryo kryo, Output out, ReplicationFactor
replicationFactor)
{
out.writeByte(replicationFactor.replicationStrategy.value);
- out.writeByte(replicationFactor.options.size());
- for (Map.Entry<String, Integer> entry :
replicationFactor.options.entrySet())
+ out.writeByte(replicationFactor.replication.size());
+ for (Map.Entry<String, ReplicaCounts> entry :
replicationFactor.replication.entrySet())
{
out.writeString(entry.getKey());
- out.writeByte(entry.getValue());
+ out.writeByte(entry.getValue().allReplicas());
+ out.writeByte(entry.getValue().transientReplicas());
Review Comment:
This adds a new field to the Kryo wire format (1 byte/DC → 2 bytes/DC) with
no version marker, so an old-format ReplicationFactor read by this new code
would crash or silently produce wrong RF values.
CassandraRing solves this exact problem for its own serialization with a
version byte, and this is actually the primary path Spark uses (its JDK
fallback is just a fallback), so it seems worth the same protection here.
##########
cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/ReplicationFactorTests.java:
##########
@@ -111,4 +118,357 @@ public void testEquality()
assertThat(replicationFactor1).isEqualTo(replicationFactor2);
assertThat(replicationFactor1.hashCode()).isEqualTo(replicationFactor2.hashCode());
}
+
+ // Transient / witness replicas: the <replicas>/<transient> form, reused
by witness replicas under
+ // mutation tracking (CEP-45/CEP-46)
+
+ @Test
+ public void testNoTransientReplicasByDefault()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3"));
+ assertThat(replicationFactor.hasTransientReplicas()).isFalse();
+ assertThat(replicationFactor.getTransientOptions()).isEmpty();
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3);
+ assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(3);
+
assertThat(replicationFactor.getTransientReplicationFactor()).isEqualTo(0);
+
assertThat(replicationFactor.getFullReplicas("datacenter1")).isEqualTo(3);
+
assertThat(replicationFactor.getTransientReplicas("datacenter1")).isEqualTo(0);
+ }
+
+ @Test
+ public void testTransientReplicasSingleDatacenter()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/1"));
+ assertThat(replicationFactor.hasTransientReplicas()).isTrue();
+ // total keeps its original meaning: all replicas, witnesses included
+
assertThat(replicationFactor.getOptions().get("datacenter1")).isEqualTo(Integer.valueOf(3));
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3);
+ assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(2);
+
assertThat(replicationFactor.getTransientReplicationFactor()).isEqualTo(1);
+
assertThat(replicationFactor.getFullReplicas("datacenter1")).isEqualTo(2);
+
assertThat(replicationFactor.getTransientReplicas("datacenter1")).isEqualTo(1);
+ }
+
+ @Test
+ public void testTransientReplicasMultipleDatacenters()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/1",
+ "datacenter2", "3/1"));
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(6);
+ assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(4);
+
assertThat(replicationFactor.getTransientReplicationFactor()).isEqualTo(2);
+ }
+
+ @Test
+ public void testMixedTransientAndFullDatacenters()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/1",
+ "datacenter2", "3"));
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(6);
+ assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(5);
+
assertThat(replicationFactor.getTransientReplicas("datacenter1")).isEqualTo(1);
+
assertThat(replicationFactor.getTransientReplicas("datacenter2")).isEqualTo(0);
+
assertThat(replicationFactor.getFullReplicas("datacenter2")).isEqualTo(3);
+ // datacenter2 has no transient replicas, so it must be absent rather
than mapped to zero
+
assertThat(replicationFactor.getTransientOptions()).containsOnlyKeys("datacenter1");
+ }
+
+ @Test
+ public void testTransientReplicasSimpleStrategy()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "SimpleStrategy",
+ "replication_factor", "3/1"));
+ assertThat(replicationFactor.getReplicationStrategy())
+ .isEqualTo(ReplicationFactor.ReplicationStrategy.SimpleStrategy);
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3);
+ assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(2);
+ }
+
+ @Test
+ public void testTransientReplicasWithWhitespace()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", " 3 / 1 "));
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3);
+ assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(2);
+ }
+
+ @Test
+ public void testTransientEqualToTotalIsRejected()
+ {
+ // Cassandra requires at least one full replica, so 3/3 is invalid and
must not be accepted silently
+ assertThatThrownBy(() -> new ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/3")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("datacenter1");
+ }
+
+ @Test
+ public void testMalformedReplicationValuesAreRejected()
+ {
+ for (String malformed : new String[]{"3/", "/1", "3/1/1", "3/x",
"x/1", "3/-1", ""})
+ {
+ assertThatThrownBy(() -> new ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", malformed)))
+ .as("malformed replication value '%s' should be rejected",
malformed)
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+ }
+
+ @Test
+ public void testGetFullReplicasUnknownDatacenter()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/1"));
+ assertThatThrownBy(() -> replicationFactor.getFullReplicas("nosuchdc"))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ public void testTransientOptionsForUnknownDatacenterRejected()
+ {
+ assertThatThrownBy(() -> new ReplicationFactor(
+ ReplicationFactor.ReplicationStrategy.NetworkTopologyStrategy,
+ ImmutableMap.of("datacenter1", 3),
+ ImmutableMap.of("datacenter2", 1)))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ public void testEqualityConsidersTransientReplicas()
+ {
+ ReplicationFactor full = new ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3"));
+ ReplicationFactor withTransient = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/1"));
+ assertThat(full).isNotEqualTo(withTransient);
+ assertThat(full.hashCode()).isNotEqualTo(withTransient.hashCode());
+ }
+
+ @Test
+ public void testNegativeReplicationFactorIsRejected()
+ {
+ assertThatThrownBy(() -> new ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "-3")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("datacenter1");
+ }
+
+ @Test
+ public void testZeroReplicationFactorIsAllowed()
+ {
+ // RF 0 is legitimate for NetworkTopologyStrategy: the keyspace is
simply not replicated to that datacenter
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3",
+ "datacenter2", "0"));
+
assertThat(replicationFactor.getOptions().get("datacenter2")).isEqualTo(Integer.valueOf(0));
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3);
+ }
+
+ @Test
+ public void testKryoSerializationRoundTripWithTransientReplicas() throws
Exception
+ {
+ ReplicationFactor original = new ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/1",
+ "datacenter2", "3"));
+
+ Kryo kryo = new Kryo();
+ kryo.register(ReplicationFactor.class, new
ReplicationFactor.Serializer());
+
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ try (Output out = new Output(bytes))
+ {
+ kryo.writeObject(out, original);
+ }
+ ReplicationFactor deserialized;
+ try (Input in = new Input(new
ByteArrayInputStream(bytes.toByteArray())))
+ {
+ deserialized = kryo.readObject(in, ReplicationFactor.class);
+ }
+
+ assertThat(deserialized).isEqualTo(original);
+ assertThat(deserialized.getTotalReplicationFactor()).isEqualTo(6);
+ assertThat(deserialized.getFullReplicationFactor()).isEqualTo(5);
+
assertThat(deserialized.getTransientReplicas("datacenter1")).isEqualTo(1);
+
assertThat(deserialized.getTransientReplicas("datacenter2")).isEqualTo(0);
+ }
+
+ @Test
+ public void testJdkSerializationRoundTripWithTransientReplicas() throws
Exception
Review Comment:
Same here, this test Is identical to
testPrecomputedValuesSurviveJdkSerialization, WDYT of merging them
##########
cassandra-analytics-common/src/test/java/org/apache/cassandra/spark/data/ReplicationFactorTests.java:
##########
@@ -111,4 +118,357 @@ public void testEquality()
assertThat(replicationFactor1).isEqualTo(replicationFactor2);
assertThat(replicationFactor1.hashCode()).isEqualTo(replicationFactor2.hashCode());
}
+
+ // Transient / witness replicas: the <replicas>/<transient> form, reused
by witness replicas under
+ // mutation tracking (CEP-45/CEP-46)
+
+ @Test
+ public void testNoTransientReplicasByDefault()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3"));
+ assertThat(replicationFactor.hasTransientReplicas()).isFalse();
+ assertThat(replicationFactor.getTransientOptions()).isEmpty();
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3);
+ assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(3);
+
assertThat(replicationFactor.getTransientReplicationFactor()).isEqualTo(0);
+
assertThat(replicationFactor.getFullReplicas("datacenter1")).isEqualTo(3);
+
assertThat(replicationFactor.getTransientReplicas("datacenter1")).isEqualTo(0);
+ }
+
+ @Test
+ public void testTransientReplicasSingleDatacenter()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/1"));
+ assertThat(replicationFactor.hasTransientReplicas()).isTrue();
+ // total keeps its original meaning: all replicas, witnesses included
+
assertThat(replicationFactor.getOptions().get("datacenter1")).isEqualTo(Integer.valueOf(3));
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3);
+ assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(2);
+
assertThat(replicationFactor.getTransientReplicationFactor()).isEqualTo(1);
+
assertThat(replicationFactor.getFullReplicas("datacenter1")).isEqualTo(2);
+
assertThat(replicationFactor.getTransientReplicas("datacenter1")).isEqualTo(1);
+ }
+
+ @Test
+ public void testTransientReplicasMultipleDatacenters()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/1",
+ "datacenter2", "3/1"));
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(6);
+ assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(4);
+
assertThat(replicationFactor.getTransientReplicationFactor()).isEqualTo(2);
+ }
+
+ @Test
+ public void testMixedTransientAndFullDatacenters()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/1",
+ "datacenter2", "3"));
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(6);
+ assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(5);
+
assertThat(replicationFactor.getTransientReplicas("datacenter1")).isEqualTo(1);
+
assertThat(replicationFactor.getTransientReplicas("datacenter2")).isEqualTo(0);
+
assertThat(replicationFactor.getFullReplicas("datacenter2")).isEqualTo(3);
+ // datacenter2 has no transient replicas, so it must be absent rather
than mapped to zero
+
assertThat(replicationFactor.getTransientOptions()).containsOnlyKeys("datacenter1");
+ }
+
+ @Test
+ public void testTransientReplicasSimpleStrategy()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "SimpleStrategy",
+ "replication_factor", "3/1"));
+ assertThat(replicationFactor.getReplicationStrategy())
+ .isEqualTo(ReplicationFactor.ReplicationStrategy.SimpleStrategy);
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3);
+ assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(2);
+ }
+
+ @Test
+ public void testTransientReplicasWithWhitespace()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", " 3 / 1 "));
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3);
+ assertThat(replicationFactor.getFullReplicationFactor()).isEqualTo(2);
+ }
+
+ @Test
+ public void testTransientEqualToTotalIsRejected()
+ {
+ // Cassandra requires at least one full replica, so 3/3 is invalid and
must not be accepted silently
+ assertThatThrownBy(() -> new ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/3")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("datacenter1");
+ }
+
+ @Test
+ public void testMalformedReplicationValuesAreRejected()
+ {
+ for (String malformed : new String[]{"3/", "/1", "3/1/1", "3/x",
"x/1", "3/-1", ""})
+ {
+ assertThatThrownBy(() -> new ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", malformed)))
+ .as("malformed replication value '%s' should be rejected",
malformed)
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+ }
+
+ @Test
+ public void testGetFullReplicasUnknownDatacenter()
+ {
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/1"));
+ assertThatThrownBy(() -> replicationFactor.getFullReplicas("nosuchdc"))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ public void testTransientOptionsForUnknownDatacenterRejected()
+ {
+ assertThatThrownBy(() -> new ReplicationFactor(
+ ReplicationFactor.ReplicationStrategy.NetworkTopologyStrategy,
+ ImmutableMap.of("datacenter1", 3),
+ ImmutableMap.of("datacenter2", 1)))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ public void testEqualityConsidersTransientReplicas()
+ {
+ ReplicationFactor full = new ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3"));
+ ReplicationFactor withTransient = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3/1"));
+ assertThat(full).isNotEqualTo(withTransient);
+ assertThat(full.hashCode()).isNotEqualTo(withTransient.hashCode());
+ }
+
+ @Test
+ public void testNegativeReplicationFactorIsRejected()
+ {
+ assertThatThrownBy(() -> new ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "-3")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("datacenter1");
+ }
+
+ @Test
+ public void testZeroReplicationFactorIsAllowed()
+ {
+ // RF 0 is legitimate for NetworkTopologyStrategy: the keyspace is
simply not replicated to that datacenter
+ ReplicationFactor replicationFactor = new
ReplicationFactor(ImmutableMap.of(
+ "class", "NetworkTopologyStrategy",
+ "datacenter1", "3",
+ "datacenter2", "0"));
+
assertThat(replicationFactor.getOptions().get("datacenter2")).isEqualTo(Integer.valueOf(0));
+ assertThat(replicationFactor.getTotalReplicationFactor()).isEqualTo(3);
+ }
+
+ @Test
+ public void testKryoSerializationRoundTripWithTransientReplicas() throws
Exception
Review Comment:
This test is almost identical to
testPrecomputedValuesSurviveKryoSerialization, WDYT of merging them?
--
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]
