support disable_read_repair_mutation and disable_block_on_read_repair

Author: rhuffy

src/java/org/apache/cassandra/config/Config.java

@@ -315,6 +315,8 @@ public class Config
     public int otc_coalescing_enough_coalesced_messages = 8;
 
     public volatile boolean coerce_read_consistency_all = false;
+    public volatile boolean disable_read_repair_mutation = false;
+    public volatile boolean disable_block_on_read_repair = false;
 
     public int windows_timer_interval = 0;
 

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -2106,4 +2106,24 @@ public static void setCoerceReadConsistencyAll(boolean value)
     {
         conf.coerce_read_consistency_all = value;
     }
+
+    public static boolean getDisableReadRepairMutation()
+    {
+        return conf.disable_read_repair_mutation;
+    }
+
+    public static void setDisableReadRepairMutation(boolean value)
+    {
+        conf.disable_read_repair_mutation = value;
+    }
+
+    public static boolean getDisableBlockOnReadRepair()
+    {
+        return conf.disable_block_on_read_repair;
+    }
+
+    public static void setDisableBlockOnReadRepair(boolean value)
+    {
+        conf.disable_block_on_read_repair = value;
+    }
 }

src/java/org/apache/cassandra/service/RowDataResolver.java

@@ -26,6 +26,7 @@
 
 import com.google.common.collect.Iterables;
 
+import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.*;
 import org.apache.cassandra.db.columniterator.IdentityQueryFilter;
 import org.apache.cassandra.db.filter.IDiskAtomFilter;
@@ -109,6 +110,11 @@ public Row resolve() throws DigestMismatchException
      */
     public static List<AsyncOneResponse> scheduleRepairs(ColumnFamily resolved, String keyspaceName, DecoratedKey key, List<ColumnFamily> versions, List<InetAddress> endpoints)
     {
+        if (DatabaseDescriptor.getDisableReadRepairMutation())
+        {
+            return Collections.emptyList();
+        }
+
         List<AsyncOneResponse> results = new ArrayList<AsyncOneResponse>(versions.size());
 
         for (int i = 0; i < versions.size(); i++)

src/java/org/apache/cassandra/service/StorageProxy.java

@@ -1556,9 +1556,12 @@ else if (logger.isDebugEnabled())
                         RowDataResolver resolver = (RowDataResolver)handler.resolver;
                         try
                         {
-                            // wait for the repair writes to be acknowledged, to minimize impact on any replica that's
-                            // behind on writes in case the out-of-sync row is read multiple times in quick succession
-                            FBUtilities.waitOnFutures(resolver.repairResults, DatabaseDescriptor.getWriteRpcTimeout());
+                            if (!DatabaseDescriptor.getDisableBlockOnReadRepair())
+                            {
+                                // wait for the repair writes to be acknowledged, to minimize impact on any replica that's
+                                // behind on writes in case the out-of-sync row is read multiple times in quick succession
+                                FBUtilities.waitOnFutures(resolver.repairResults, DatabaseDescriptor.getWriteRpcTimeout());
+                            }
                         }
                         catch (TimeoutException e)
                         {

src/java/org/apache/cassandra/service/StorageService.java

@@ -5528,14 +5528,36 @@ public void setWriteDelay(int writeDelay) {
         logger.info("Updated write_delay_in_s to {}", SafeArg.of("value", writeDelay));
     }
 
+    public boolean getCoerceReadConsistencyAll()
+    {
+        return DatabaseDescriptor.getCoerceReadConsistencyAll();
+    }
+
     public void setCoerceReadConsistencyAll(boolean value)
     {
         DatabaseDescriptor.setCoerceReadConsistencyAll(value);
         logger.info("Updated coerce_read_consistency_level to {}", SafeArg.of("value", value));
     }
 
-    public boolean getCoerceReadConsistencyAll()
+    public boolean getDisableReadRepairMutation()
     {
-        return DatabaseDescriptor.getCoerceReadConsistencyAll();
+        return DatabaseDescriptor.getDisableReadRepairMutation();
+    }
+
+    public void setDisableReadRepairMutation(boolean value)
+    {
+        DatabaseDescriptor.setDisableReadRepairMutation(value);
+        logger.info("Updated disable_read_repair_mutation to {}", SafeArg.of("value", value));
+    }
+
+    public boolean getDisableBlockOnReadRepair()
+    {
+        return DatabaseDescriptor.getDisableBlockOnReadRepair();
+    }
+
+    public void setDisableBlockOnReadRepair(boolean value)
+    {
+        DatabaseDescriptor.setDisableBlockOnReadRepair(value);
+        logger.info("Updated disable_block_on_read_repair to {}", SafeArg.of("value", value));
     }
 }

src/java/org/apache/cassandra/service/StorageServiceMBean.java

@@ -936,4 +936,12 @@ public enum ProgressState
     public boolean getCoerceReadConsistencyAll();
 
     public void setCoerceReadConsistencyAll(boolean value);
+
+    public boolean getDisableReadRepairMutation();
+
+    public void setDisableReadRepairMutation(boolean value);
+
+    public boolean getDisableBlockOnReadRepair();
+
+    public void setDisableBlockOnReadRepair(boolean value);
 }

src/java/org/apache/cassandra/tools/NodeProbe.java

@@ -1194,6 +1194,26 @@ public void setCoerceReadConsistencyAll(boolean value)
         ssProxy.setCoerceReadConsistencyAll(value);
     }
 
+    public boolean getDisableReadRepairMutation()
+    {
+        return ssProxy.getDisableReadRepairMutation();
+    }
+
+    public void setDisableReadRepairMutation(boolean value)
+    {
+        ssProxy.setDisableReadRepairMutation(value);
+    }
+
+    public boolean getDisableBlockOnReadRepair()
+    {
+        return ssProxy.getDisableBlockOnReadRepair();
+    }
+
+    public void setDisableBlockOnReadRepair(boolean value)
+    {
+        ssProxy.setDisableBlockOnReadRepair(value);
+    }
+
     // JMX getters for the o.a.c.metrics API below.
     /**
      * Retrieve cache metrics based on the cache type (KeyCache, RowCache, or CounterCache)

src/java/org/apache/cassandra/tools/nodetool/GetDisableBlockOnReadRepair.java

@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.tools.nodetool;
+
+import io.airlift.command.Command;
+import org.apache.cassandra.tools.NodeProbe;
+import org.apache.cassandra.tools.NodeTool;
+
+@Command(name = "getdisableblockonreadrepair", description = "Check if blocking on read repair is disabled")
+public class GetDisableBlockOnReadRepair extends NodeTool.NodeToolCmd
+{
+    @Override
+    public void execute(NodeProbe probe)
+    {
+        probe.output().out.println(probe.getDisableBlockOnReadRepair());
+    }
+}

src/java/org/apache/cassandra/tools/nodetool/GetDisableReadRepairMutation.java

@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.tools.nodetool;
+
+import io.airlift.command.Command;
+import org.apache.cassandra.tools.NodeProbe;
+import org.apache.cassandra.tools.NodeTool;
+
+@Command(name = "getdisablereadrepairmutation", description = "Check if read repair mutations are disabled")
+public class GetDisableReadRepairMutation extends NodeTool.NodeToolCmd
+{
+    @Override
+    public void execute(NodeProbe probe)
+    {
+        probe.output().out.println(probe.getDisableReadRepairMutation());
+    }
+}

src/java/org/apache/cassandra/tools/nodetool/SetDisableBlockOnReadRepair.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.tools.nodetool;
+
+import io.airlift.command.Arguments;
+import io.airlift.command.Command;
+import org.apache.cassandra.tools.NodeProbe;
+import org.apache.cassandra.tools.NodeTool;
+
+@Command(name = "setdisableblockonreadrepair", description = "Disable block on read repair")
+public class SetDisableBlockOnReadRepair extends NodeTool.NodeToolCmd
+{
+    @Arguments(title = "flag", description = "true, or false (default)", required = true)
+    private String flag;
+
+    @Override
+    public void execute(NodeProbe probe)
+    {
+        probe.setDisableBlockOnReadRepair(Boolean.parseBoolean(flag));
+    }
+}

src/java/org/apache/cassandra/tools/nodetool/SetDisableReadRepairMutation.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.tools.nodetool;
+
+import io.airlift.command.Arguments;
+import io.airlift.command.Command;
+import org.apache.cassandra.tools.NodeProbe;
+import org.apache.cassandra.tools.NodeTool;
+
+@Command(name = "setdisablereadrepairmutation", description = "Disable read repair mutations")
+public class SetDisableReadRepairMutation extends NodeTool.NodeToolCmd
+{
+    @Arguments(title = "flag", description = "true, or false (default)", required = true)
+    private String flag;
+
+    @Override
+    public void execute(NodeProbe probe)
+    {
+        probe.setDisableReadRepairMutation(Boolean.parseBoolean(flag));
+    }
+}