throw an exception when serving an invalid read

src/java/com/palantir/cassandra/utils/MutationVerificationHandler.java

@@ -21,6 +21,7 @@
 import com.palantir.logsafe.SafeArg;
 import com.palantir.logsafe.UnsafeArg;
 import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.Mutation;
 import org.apache.cassandra.exceptions.InvalidMutationException;
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.Hex;
@@ -32,20 +33,20 @@
 import java.util.Collection;
 import java.util.List;
 
-public class MutationVerificationHandler implements OwnershipVerificationHandler
+public class MutationVerificationHandler implements OwnershipVerificationHandler<Mutation>
 {
     private static final Logger logger = LoggerFactory.getLogger(OwnershipVerificationUtils.class);
 
-    public static final OwnershipVerificationHandler INSTANCE = new MutationVerificationHandler();
+    public static final OwnershipVerificationHandler<Mutation> INSTANCE = new MutationVerificationHandler();
 
     @Override
-    public void onViolation(Keyspace keyspace, ByteBuffer key, List<InetAddress> naturalEndpoints, Collection<InetAddress> pendingEndpoints)
+    public void onViolation(Mutation mutation, Keyspace keyspace, List<InetAddress> naturalEndpoints, Collection<InetAddress> pendingEndpoints)
     {
         keyspace.metric.invalidMutations.inc();
         logger.error(
             "InvalidMutation! Cannot apply mutation as this host {} does not contain key {} in keyspace {}. Only hosts {} and {} do.",
             SafeArg.of("address", FBUtilities.getBroadcastAddress()),
-            UnsafeArg.of("key", Hex.bytesToHex(key.array())),
+            UnsafeArg.of("key", Hex.bytesToHex(mutation.key().array())),
             SafeArg.of("keyspace", keyspace.getName()),
             SafeArg.of("naturalEndpoints", naturalEndpoints),
             SafeArg.of("pendingEndpoints", pendingEndpoints));

src/java/com/palantir/cassandra/utils/OwnershipVerificationHandler.java

@@ -21,13 +21,12 @@
 import org.apache.cassandra.db.Keyspace;
 
 import java.net.InetAddress;
-import java.nio.ByteBuffer;
 import java.util.Collection;
 import java.util.List;
 
-public interface OwnershipVerificationHandler
+public interface OwnershipVerificationHandler<T>
 {
-    void onViolation(Keyspace keyspace, ByteBuffer key, List<InetAddress> naturalEndpoints, Collection<InetAddress> pendingEndpoints);
+    void onViolation(T payload, Keyspace keyspace, List<InetAddress> naturalEndpoints, Collection<InetAddress> pendingEndpoints);
 
     void onValid(Keyspace keyspace);
 }

src/java/com/palantir/cassandra/utils/OwnershipVerificationUtils.java

@@ -31,8 +31,10 @@
 
 import com.palantir.logsafe.SafeArg;
 import com.palantir.logsafe.UnsafeArg;
+import org.apache.cassandra.db.AbstractRangeCommand;
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.db.Mutation;
+import org.apache.cassandra.db.ReadCommand;
 import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.locator.NetworkTopologyStrategy;
 import org.apache.cassandra.service.StorageService;
@@ -43,6 +45,7 @@ public class OwnershipVerificationUtils
 {
     private static final boolean VERIFY_KEYS_ON_WRITE = Boolean.getBoolean("palantir_cassandra.verify_keys_on_write");
     private static final boolean VERIFY_KEYS_ON_READ = Boolean.getBoolean("palantir_cassandra.verify_keys_on_read");
+    private static final boolean VERIFY_KEYS_ON_RANGE_SLICE = Boolean.getBoolean("palantir_cassandra.verify_keys_on_range_slice");
     private static final Logger logger = LoggerFactory.getLogger(OwnershipVerificationUtils.class);
 
     private static volatile Instant lastTokenRingCacheUpdate = Instant.MIN;
@@ -51,41 +54,64 @@ private OwnershipVerificationUtils()
     {
     }
 
-    public static void verifyRead(Keyspace keyspace, ByteBuffer key)
+    public static void verifyMutation(Mutation mutation)
+    {
+        if (!VERIFY_KEYS_ON_WRITE)
+        {
+            return;
+        }
+        verifyOperation(
+            mutation,
+            Keyspace.open(mutation.getKeyspaceName()),
+            StorageService.getPartitioner().getToken(mutation.key()),
+            Hex.bytesToHex(mutation.key().array()),
+            MutationVerificationHandler.INSTANCE);
+    }
+
+    public static void verifyRead(ReadCommand command)
     {
         if (!VERIFY_KEYS_ON_READ)
         {
             return;
         }
-        verifyOperation(keyspace, key, ReadVerificationHandler.INSTANCE);
+        verifyOperation(
+            command,
+            Keyspace.open(command.getKeyspace()),
+            StorageService.getPartitioner().getToken(command.key),
+            Hex.bytesToHex(command.key.array()),
+            ReadVerificationHandler.INSTANCE);
     }
 
-    public static void verifyMutation(Mutation mutation)
+    public static void verifyRangeSlice(AbstractRangeCommand command)
     {
-        if (!VERIFY_KEYS_ON_WRITE)
+        if (!VERIFY_KEYS_ON_RANGE_SLICE)
         {
             return;
         }
-        verifyOperation(Keyspace.open(mutation.getKeyspaceName()), mutation.key(), MutationVerificationHandler.INSTANCE);
+        verifyOperation(
+            command,
+            Keyspace.open(command.keyspace),
+            command.keyRange.right.getToken(),
+            command.keyRange.right.toString(),
+            RangeSliceVerificationHandler.INSTANCE);
     }
 
-    private static void verifyOperation(Keyspace keyspace, ByteBuffer key, OwnershipVerificationHandler handler)
+    private static <T> void verifyOperation(T payload, Keyspace keyspace, Token tk, String keyToLog, OwnershipVerificationHandler<T> handler)
     {
         if (!(keyspace.getReplicationStrategy() instanceof NetworkTopologyStrategy))
         {
             return;
         }
 
         String keyspaceName = keyspace.getName();
-        Token tk = StorageService.getPartitioner().getToken(key);
         List<InetAddress> cachedNaturalEndpoints = StorageService.instance.getNaturalEndpoints(keyspaceName, tk);
         Collection<InetAddress> pendingEndpoints = StorageService.instance.getTokenMetadata().pendingEndpointsFor(tk, keyspaceName);
 
         if (operationIsInvalid(cachedNaturalEndpoints, pendingEndpoints))
         {
             if (cacheWasRecentlyRefreshed())
             {
-                handler.onViolation(keyspace, key, cachedNaturalEndpoints, pendingEndpoints);
+                handler.onViolation(payload, keyspace, cachedNaturalEndpoints, pendingEndpoints);
                 return;
             }
 
@@ -95,14 +121,14 @@ private static void verifyOperation(Keyspace keyspace, ByteBuffer key, Ownership
 
             if (operationIsInvalid(refreshedNaturalEndpoints, pendingEndpoints))
             {
-                handler.onViolation(keyspace, key, refreshedNaturalEndpoints, pendingEndpoints);
+                handler.onViolation(payload, keyspace, refreshedNaturalEndpoints, pendingEndpoints);
                 return;
             }
             else
             {
                 logger.warn("Ignoring InvalidOwnership error detected using stale token ring cache. Error was originally detected for key {} in keyspace {}."
                                 + " Cached owners {}. Actual owners {}. Pending owners (non-cached) {}.",
-                            UnsafeArg.of("key", Hex.bytesToHex(key.array())),
+                            UnsafeArg.of("key", keyToLog),
                             SafeArg.of("keyspace", keyspaceName),
                             SafeArg.of("cachedNaturalEndpoints", cachedNaturalEndpoints),
                             SafeArg.of("refreshedNaturalEndpoints", refreshedNaturalEndpoints),

src/java/com/palantir/cassandra/utils/RangeSliceVerificationHandler.java

@@ -0,0 +1,64 @@
+/*
+ * 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 com.palantir.cassandra.utils;
+
+import java.net.InetAddress;
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.palantir.logsafe.SafeArg;
+import com.palantir.logsafe.UnsafeArg;
+import org.apache.cassandra.db.AbstractRangeCommand;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.RangeSliceCommand;
+import org.apache.cassandra.exceptions.InvalidReadException;
+import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.Hex;
+
+public class RangeSliceVerificationHandler implements OwnershipVerificationHandler<AbstractRangeCommand>
+{
+    private static final Logger logger = LoggerFactory.getLogger(OwnershipVerificationUtils.class);
+
+    public static final OwnershipVerificationHandler<AbstractRangeCommand> INSTANCE = new RangeSliceVerificationHandler();
+
+    @Override
+    public void onViolation(AbstractRangeCommand command, Keyspace keyspace, List<InetAddress> naturalEndpoints, Collection<InetAddress> pendingEndpoints)
+    {
+        keyspace.metric.invalidRangeSlice.inc();
+        logger.error(
+        "Received Invalid RangeSlice request! This host {} does not contain range ({}, {}) in keyspace {}. Only hosts {} and {} do.",
+            SafeArg.of("address", FBUtilities.getBroadcastAddress()),
+            UnsafeArg.of("left", command.keyRange.left),
+            UnsafeArg.of("right", command.keyRange.right),
+            SafeArg.of("keyspace", keyspace.getName()),
+            SafeArg.of("naturalEndpoints", naturalEndpoints),
+            SafeArg.of("pendingEndpoints", pendingEndpoints));
+        throw new InvalidReadException();
+    }
+
+    @Override
+    public void onValid(Keyspace keyspace)
+    {
+        keyspace.metric.validRangeSlice.inc();
+    }
+}

src/java/com/palantir/cassandra/utils/ReadVerificationHandler.java

@@ -21,6 +21,8 @@
 import com.palantir.logsafe.SafeArg;
 import com.palantir.logsafe.UnsafeArg;
 import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.ReadCommand;
+import org.apache.cassandra.exceptions.InvalidReadException;
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.Hex;
 import org.slf4j.Logger;
@@ -31,23 +33,24 @@
 import java.util.Collection;
 import java.util.List;
 
-public class ReadVerificationHandler implements OwnershipVerificationHandler
+public class ReadVerificationHandler implements OwnershipVerificationHandler<ReadCommand>
 {
     private static final Logger logger = LoggerFactory.getLogger(OwnershipVerificationUtils.class);
 
-    public static final OwnershipVerificationHandler INSTANCE = new ReadVerificationHandler();
+    public static final OwnershipVerificationHandler<ReadCommand> INSTANCE = new ReadVerificationHandler();
 
     @Override
-    public void onViolation(Keyspace keyspace, ByteBuffer key, List<InetAddress> naturalEndpoints, Collection<InetAddress> pendingEndpoints)
+    public void onViolation(ReadCommand payload, Keyspace keyspace, List<InetAddress> naturalEndpoints, Collection<InetAddress> pendingEndpoints)
     {
         keyspace.metric.invalidReads.inc();
         logger.error(
-            "Executed InvalidRead! This host {} does not contain key {} in keyspace {}. Only hosts {} and {} do.",
+        "Received InvalidRead request! This host {} does not contain key {} in keyspace {}. Only hosts {} and {} do.",
             SafeArg.of("address", FBUtilities.getBroadcastAddress()),
-            UnsafeArg.of("key", Hex.bytesToHex(key.array())),
+            UnsafeArg.of("key", Hex.bytesToHex(payload.key.array())),
             SafeArg.of("keyspace", keyspace.getName()),
             SafeArg.of("naturalEndpoints", naturalEndpoints),
             SafeArg.of("pendingEndpoints", pendingEndpoints));
+        throw new InvalidReadException();
     }
 
     @Override

src/java/org/apache/cassandra/db/ReadVerbHandler.java

@@ -42,6 +42,7 @@ public void doVerb(MessageIn<ReadCommand> message, int id)
         }
 
         ReadCommand command = message.payload;
+        OwnershipVerificationUtils.verifyRead(command);
         Keyspace keyspace = Keyspace.open(command.ksName);
         Row row = command.getRow(keyspace);
 
@@ -50,8 +51,6 @@ public void doVerb(MessageIn<ReadCommand> message, int id)
                                                                       ReadResponse.serializer);
         Tracing.trace("Enqueuing response to {}", message.from);
         MessagingService.instance().sendReply(reply, id, message.from);
-
-        OwnershipVerificationUtils.verifyRead(keyspace, command.key);
     }
 
     public static ReadResponse getResponse(ReadCommand command, Row row)

src/java/org/apache/cassandra/exceptions/InvalidReadException.java

@@ -0,0 +1,27 @@
+/*
+ * 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.exceptions;
+
+public class InvalidReadException extends RuntimeException
+{
+    public InvalidReadException()
+    {
+        super("InvalidRead! Cannot serve this read as this host does not contain key.");
+    }
+}

src/java/org/apache/cassandra/metrics/KeyspaceMetrics.java

@@ -120,7 +120,11 @@ public class KeyspaceMetrics
     public final Counter invalidReads;
     /** Number of read requests which are valid */
     public final Counter validReads;
-
+    /** Number of read requests which were not for a row this node owned */
+    public final Counter invalidRangeSlice;
+    /** Number of read requests which are valid */
+    public final Counter validRangeSlice;
+
     public final MetricNameFactory factory;
     private Keyspace keyspace;
 
@@ -279,6 +283,9 @@ public Long getValue(ColumnFamilyMetrics metric)
         validMutations = Metrics.counter(factory.createMetricName("ValidMutations"));
         invalidReads = Metrics.counter(factory.createMetricName("InvalidReads"));
         validReads = Metrics.counter(factory.createMetricName("ValidReads"));
+        invalidRangeSlice = Metrics.counter(factory.createMetricName("ValidRangeSlice"));
+        validRangeSlice = Metrics.counter(factory.createMetricName("InvalidRangeSlice"));
+
     }
 
     /**

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

@@ -17,6 +17,7 @@
  */
 package org.apache.cassandra.service;
 
+import com.palantir.cassandra.utils.OwnershipVerificationUtils;
 import org.apache.cassandra.db.AbstractRangeCommand;
 import org.apache.cassandra.db.RangeSliceReply;
 import org.apache.cassandra.exceptions.IsBootstrappingException;
@@ -34,6 +35,7 @@ public void doVerb(MessageIn<AbstractRangeCommand> message, int id)
             /* Don't service reads! */
             throw new IsBootstrappingException();
         }
+        OwnershipVerificationUtils.verifyRangeSlice(message.payload);
         RangeSliceReply reply = new RangeSliceReply(message.payload.executeLocally());
         Tracing.trace("Enqueuing response to {}", message.from);
         MessagingService.instance().sendReply(reply.createMessage(), id, message.from);