CNDB-15667: CNDB-15623: Only use write path for CDC tables in CassandraStreamReceiver if CDC is enabled on the node (#2043)

Repairs use the local write path for streams on CDC-enabled tables,
based on table schema. This interacts poorly with the separation of CNDB
services.

This commit fixes the issue by only using the CDC write path for a stream if CDC
is enabled in the node's configuration (as well as in the schema). This avoids
attempting to use the local write path if commitlog-based CDC is not enabled.

Author: jkni

src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java

@@ -182,7 +182,7 @@ private boolean hasViews(ColumnFamilyStore cfs)
 
     private boolean hasCDC(ColumnFamilyStore cfs)
     {
-        return cfs.metadata().params.cdc;
+        return DatabaseDescriptor.isCDCEnabled() && cfs.metadata().params.cdc;
     }
 
     // returns true iif it is a cdc table and cdc on repair is enabled.

test/distributed/org/apache/cassandra/distributed/test/RepairTest.java

@@ -29,19 +29,25 @@
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.distributed.shared.ClusterUtils;
 import org.apache.cassandra.utils.concurrent.Condition;
+
 import org.junit.AfterClass;
 import org.junit.Assert;
-import org.junit.BeforeClass;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 
 import org.apache.cassandra.distributed.Cluster;
 import org.apache.cassandra.distributed.api.ICluster;
 import org.apache.cassandra.distributed.api.IInstanceConfig;
 import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.schema.SchemaConstants;
+import org.apache.cassandra.schema.SystemDistributedKeyspace;
 import org.apache.cassandra.service.StorageService;
 
 import static com.google.common.collect.ImmutableList.of;
+
 import static java.util.concurrent.TimeUnit.MINUTES;
+
 import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
 import static org.apache.cassandra.distributed.api.Feature.NETWORK;
 import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows;
@@ -50,10 +56,19 @@
 import static org.apache.cassandra.utils.concurrent.Condition.newOneTimeCondition;
 import static org.apache.cassandra.utils.progress.ProgressEventType.COMPLETE;
 
+@RunWith(Parameterized.class)
 public class RepairTest extends TestBaseImpl
 {
+    private static boolean nodesHaveCDC;
+    private static boolean tableHasCDC;
     private static ICluster<IInvokableInstance> cluster;
 
+    @Parameterized.Parameters(name = "nodesHaveCDC={0}, tableHasCDC={1}")
+    public static Iterable<Object[]> data()
+    {
+        return Arrays.asList(new Object[][] {{ false, false }, { false, true } , { true, false }, { true, true }});
+    }
+
     private static void insert(ICluster<IInvokableInstance> cluster, String keyspace, int start, int end, int ... nodes)
     {
         String insert = String.format("INSERT INTO %s.test (k, c1, c2) VALUES (?, 'value1', 'value2');", keyspace);
@@ -85,7 +100,7 @@ private static void flush(ICluster<IInvokableInstance> cluster, String keyspace,
                                                                                                      ColumnFamilyStore.FlushReason.UNIT_TESTS)));
     }
 
-    private static ICluster create(Consumer<IInstanceConfig> configModifier) throws IOException
+    private ICluster create(Consumer<IInstanceConfig> configModifier) throws IOException
     {
         configModifier = configModifier.andThen(
         config -> config.set("hinted_handoff_enabled", false)
@@ -98,6 +113,15 @@ private static ICluster create(Consumer<IInstanceConfig> configModifier) throws
 
     static void repair(ICluster<IInvokableInstance> cluster, String keyspace, Map<String, String> options)
     {
+        long[] startPositions = new long[cluster.size()];
+        for (int i = 1; i <= cluster.size(); i++)
+        {
+            IInvokableInstance node = cluster.get(i);
+            if (node.isShutdown())
+                continue;
+            startPositions[i - 1] = node.logs().mark();
+        }
+
         cluster.get(1).runOnInstance(rethrow(() -> {
             Condition await = newOneTimeCondition();
             instance.repair(keyspace, options, of((tag, event) -> {
@@ -106,14 +130,25 @@ static void repair(ICluster<IInvokableInstance> cluster, String keyspace, Map<St
             })).right.get();
             await.await(1L, MINUTES);
         }));
+
+        for (int i = 1; i <= cluster.size(); i++)
+        {
+            IInvokableInstance node = cluster.get(i);
+            if (node.isShutdown())
+                continue;
+            Assert.assertEquals("We should use the local write path (which requires flushing) if CDC is enabled on both a node and table-level",
+                                nodesHaveCDC && tableHasCDC,
+                                !node.logs().grep(startPositions[i - 1],
+                                                 String.format("Enqueuing flush of %s.test, Reason\\: STREAMS_RECEIVED", keyspace)).getResult().isEmpty());
+        }
     }
 
-    static void populate(ICluster<IInvokableInstance> cluster, String keyspace, String compression) throws Exception
+    void populate(ICluster<IInvokableInstance> cluster, String keyspace, String compression) throws Exception
     {
         try
         {
             cluster.schemaChange(String.format("DROP TABLE IF EXISTS %s.test;", keyspace));
-            cluster.schemaChange(String.format("CREATE TABLE %s.test (k text, c1 text, c2 text, PRIMARY KEY (k)) WITH compression = %s", keyspace, compression));
+            cluster.schemaChange(String.format("CREATE TABLE %s.test (k text, c1 text, c2 text, PRIMARY KEY (k)) WITH compression = %s AND cdc = %s;", keyspace, compression, tableHasCDC));
 
             insert(cluster, keyspace, 0, 1000, 1, 2, 3);
             flush(cluster, keyspace, 1);
@@ -147,10 +182,21 @@ void shutDownNodesAndForceRepair(ICluster<IInvokableInstance> cluster, String ke
         repair(cluster, keyspace, ImmutableMap.of("forceRepair", "true"));
     }
 
-    @BeforeClass
-    public static void setupCluster() throws IOException
+    public RepairTest(boolean nodesHaveCDC, boolean tableHasCDC) throws Exception
     {
-        cluster = create(config -> {});
+        // This runs per method, but we only want to rebuild the cluster if nodesHaveCDC has changed since the last
+        // build and we need to update the configuration accordingly
+        if (cluster != null && RepairTest.nodesHaveCDC != nodesHaveCDC)
+        {
+            cluster.close();
+            cluster = null;
+        }
+
+        if (cluster == null)
+            cluster = create(config -> config.set("cdc_enabled", nodesHaveCDC));
+
+        RepairTest.nodesHaveCDC = nodesHaveCDC;
+        RepairTest.tableHasCDC = tableHasCDC;
     }
 
     @AfterClass
@@ -203,7 +249,12 @@ public void testForcedNormalRepairWithOneNodeDown() throws Exception
         String forceRepairKeyspace = "test_force_repair_keyspace";
         int rf = 2;
         int tokenCount = ClusterUtils.getTokenCount(cluster.get(1));
-        cluster.schemaChange("CREATE KEYSPACE " + forceRepairKeyspace + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': " + rf + "};");
+
+        cluster.schemaChange("CREATE KEYSPACE IF NOT EXISTS " + forceRepairKeyspace + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': " + rf + "};");
+
+        // Truncate distributed repair keyspace due to test class parameterization. We only want results
+        // from our run
+        cluster.schemaChange("TRUNCATE TABLE " + SchemaConstants.DISTRIBUTED_KEYSPACE_NAME + "." + SystemDistributedKeyspace.PARENT_REPAIR_HISTORY);
 
         try
         {