CASSANALYTICS-194: Parse <replicas>/<transient> replication factor fo…

CHANGES.txt

@@ -1,5 +1,6 @@
 0.5.0
 -----
+ * Parse <replicas>/<transient> replication factor for witness-enabled keyspaces (CASSANALYTICS-194)
  * Determine whether mutation tracking is enabled for keyspace for bulk writes (CASSANALYTICS-160)
  * Upgrade sidecar version to 0.4.0
  * CDC logs NPE for deleted column values (CASSANALYTICS-178)

cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/partitioner/CassandraRing.java

@@ -28,9 +28,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.stream.Collectors;
 
 import java.util.Objects;
@@ -69,6 +67,20 @@ public class CassandraRing implements Serializable
     private static final Logger LOGGER = LoggerFactory.getLogger(CassandraRing.class);
     public static final Serializer SERIALIZER = new Serializer();
 
+    /**
+     * Pinned so that the JDK serialization format change made when transient (witness) replica counts were added is
+     * detected. Without it the UID is computed from the class signature, which did not change - only the
+     * {@link #readObject}/{@link #writeObject} bodies did - so an older stream would be silently misread rather than
+     * rejected.
+     */
+    private static final long serialVersionUID = 2026082800000000001L;
+
+    /**
+     * Incremented whenever the hand-rolled JDK serialization format below changes. Version 1 writes the
+     * ReplicationFactor as an object rather than destructuring it.
+     */
+    private static final byte SERIALIZATION_FORMAT_VERSION = 1;
+
     private Partitioner partitioner;
     private String keyspace;
     private ReplicationFactor replicationFactor;
@@ -260,17 +272,18 @@ public int hashCode()
     private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException
     {
         LOGGER.debug("Falling back to JDK deserialization");
+        byte formatVersion = in.readByte();
+        if (formatVersion != SERIALIZATION_FORMAT_VERSION)
+        {
+            throw new IOException(String.format("Unsupported CassandraRing serialization format version %d, expected %d",
+                                                formatVersion, SERIALIZATION_FORMAT_VERSION));
+        }
         this.partitioner = in.readByte() == 0 ? Partitioner.RandomPartitioner : Partitioner.Murmur3Partitioner;
         this.keyspace = in.readUTF();
 
-        ReplicationFactor.ReplicationStrategy strategy = ReplicationFactor.ReplicationStrategy.valueOf(in.readByte());
-        int optionCount = in.readByte();
-        Map<String, Integer> options = new HashMap<>(optionCount);
-        for (int option = 0; option < optionCount; option++)
-        {
-            options.put(in.readUTF(), (int) in.readByte());
-        }
-        this.replicationFactor = new ReplicationFactor(strategy, options);
+        // ReplicationFactor is Serializable, so it is written whole rather than destructured. That keeps this
+        // method independent of how ReplicationFactor represents its per-datacenter counts.
+        this.replicationFactor = (ReplicationFactor) in.readObject();
 
         int numInstances = in.readShort();
         this.instances = new ArrayList<>(numInstances);
@@ -284,17 +297,11 @@ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundE
     private void writeObject(ObjectOutputStream out) throws IOException, ClassNotFoundException
     {
         LOGGER.debug("Falling back to JDK serialization");
+        out.writeByte(SERIALIZATION_FORMAT_VERSION);
         out.writeByte(this.partitioner == Partitioner.RandomPartitioner ? 0 : 1);
         out.writeUTF(this.keyspace);
 
-        out.writeByte(this.replicationFactor.getReplicationStrategy().value);
-        Map<String, Integer> options = this.replicationFactor.getOptions();
-        out.writeByte(options.size());
-        for (Map.Entry<String, Integer> option : options.entrySet())
-        {
-            out.writeUTF(option.getKey());
-            out.writeByte(option.getValue());
-        }
+        out.writeObject(this.replicationFactor);
 
         out.writeShort(this.instances.size());
         for (CassandraInstance instance : this.instances)

cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java

@@ -176,9 +176,18 @@ public static ReplicationFactor extractReplicationFactor(@NotNull String schemaS
             throw new RuntimeException(String.format("Unable to parse replication factor for keyspace: %s", keyspace), exception);
         }
 
-        String className = map.remove("class");
-        ReplicationFactor.ReplicationStrategy strategy = ReplicationFactor.ReplicationStrategy.getEnum(className);
-        return new ReplicationFactor(strategy, map.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, v -> Integer.parseInt(v.getValue()))));
+        // Values may use the <replicas>/<transient> form for witness replicas, so delegate parsing to
+        // ReplicationFactor, which reports an unparseable value rather than dropping the datacenter. Dropping it
+        // would surface later as a confusing "DC not found in replication factor" error.
+        try
+        {
+            return new ReplicationFactor(map);
+        }
+        catch (IllegalArgumentException exception)
+        {
+            throw new RuntimeException(String.format("Unable to parse replication factor for keyspace: %s", keyspace),
+                                       exception);
+        }
     }
 
     public static String extractTableSchema(@NotNull String schemaStr, @NotNull String keyspace, @NotNull String table)