mansikhara commented on code in PR #238:
URL:
https://github.com/apache/cassandra-analytics/pull/238#discussion_r4086134291
##########
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:
Agreed, done!
##########
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:
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]
