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

CHANGES.txt

@@ -1,5 +1,7 @@
 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
  * Exclude IP address from RingInstance equality so node replacement does not fail bulk write jobs (CASSANALYTICS-175)
  * Regenerate bloom filters for CQLSSTableWriter (CASSANALYTICS-167)

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

@@ -69,6 +69,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 +274,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 +299,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

@@ -52,6 +52,9 @@ public final class CqlUtils
                                                                                         "max_index_interval"
                                                                                         );
     private static final Pattern REPLICATION_FACTOR_PATTERN = Pattern.compile("WITH REPLICATION = (\\{[^\\}]*\\})");
+    private static final String TRACKED_REPLICATION_TYPE = "tracked";
+    private static final Pattern REPLICATION_TYPE_PATTERN = Pattern.compile("replication_type\\s*=\\s*'(\\w+)'",
+                                                                             Pattern.CASE_INSENSITIVE);
     // Initialize a mapper allowing single quotes to process the RF string from the CREATE KEYSPACE statement
     private static final ObjectMapper MAPPER = new ObjectMapper().configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true);
     private static final Pattern ESCAPED_WHITESPACE_PATTERN = Pattern.compile("(\\\\r|\\\\n|\\\\r\\n)+");
@@ -173,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)
@@ -296,4 +308,33 @@ public static boolean isTimeRangeFilterSupported(String compactionStrategy)
     {
         return compactionStrategy == null || compactionStrategy.endsWith("TimeWindowCompactionStrategy");
     }
+
+    /**
+     * Extracts replication type from create schema statement
+     *
+     * @param schemaStr full cluster schema string as returned by Sidecar
+     * @param keyspace  name of the keyspace to check
+     * @return {@code true} if keyspace is tracked {@code false} otherwise
+     */
+    public static String extractReplicationType(@NotNull String schemaStr, @NotNull String keyspace)
+    {
+        String createKeyspaceSchema = extractKeyspaceSchema(schemaStr, keyspace);
+        Matcher matcher = REPLICATION_TYPE_PATTERN.matcher(createKeyspaceSchema);
+        if (matcher.find())
+        {
+            return matcher.group(1);
+        }
+        return null;
+    }
+
+    /**
+     * Returns {@code true} if {@code replication_type = 'tracked'} in create statement otherwise {@code false}
+     *
+     * @param replicationType replication type extracted from create statement
+     * @return {@code true} if replication type is tracked {@code false} otherwise
+     */
+    public static boolean isTracked(String replicationType)
+    {
+        return TRACKED_REPLICATION_TYPE.equalsIgnoreCase(replicationType);
+    }
 }