Merge branch 'cassandra-3.0' into trunk

src/java/org/apache/cassandra/net/WriteCallbackInfo.java

@@ -28,9 +28,8 @@ import org.apache.cassandra.service.paxos.Commit;
 
 public class WriteCallbackInfo extends CallbackInfo
 {
-    private final MessageOut sentMessage;
-    private final ConsistencyLevel consistencyLevel;
-    private final boolean allowHints;
+    // either a Mutation, or a Paxos Commit (MessageOut)
+    private final Object mutation;
 
     public WriteCallbackInfo(InetAddress target,
                              IAsyncCallback callback,
@@ -41,23 +40,32 @@ public class WriteCallbackInfo extends CallbackInfo
     {
         super(target, callback, serializer, true);
         assert message != null;
-        this.sentMessage = message;
-        this.consistencyLevel = consistencyLevel;
-        this.allowHints = allowHints;
-    }
-
-    Mutation mutation()
-    {
-        return sentMessage.verb == MessagingService.Verb.PAXOS_COMMIT
-             ? ((Commit) sentMessage.payload).makeMutation()
-             : (Mutation) sentMessage.payload;
+        this.mutation = shouldHint(allowHints, message, consistencyLevel);
     }
 
     public boolean shouldHint()
     {
-        return allowHints
-            && sentMessage.verb != MessagingService.Verb.COUNTER_MUTATION
-            && consistencyLevel != ConsistencyLevel.ANY
-            && StorageProxy.shouldHint(target);
+        return mutation != null && StorageProxy.shouldHint(target);
     }
+
+    public Mutation mutation()
+    {
+        return getMutation(mutation);
+    }
+
+    private static Mutation getMutation(Object object)
+    {
+        assert object instanceof Commit || object instanceof Mutation : object;
+        return object instanceof Commit ? ((Commit) object).makeMutation()
+                                        : (Mutation) object;
+    }
+
+    private static Object shouldHint(boolean allowHints, MessageOut sentMessage, ConsistencyLevel consistencyLevel)
+    {
+        return allowHints
+               && sentMessage.verb != MessagingService.Verb.COUNTER_MUTATION
+               && consistencyLevel != ConsistencyLevel.ANY
+               ? sentMessage.payload : null;
+    }
+
 }

test/unit/org/apache/cassandra/MockSchema.java

@@ -142,7 +142,7 @@ public class MockSchema
         return new ColumnFamilyStore(ks, cfname, 0, metadata, new Directories(metadata), false, false);
     }
 
-    private static CFMetaData newCFMetaData(String ksname, String cfname)
+    public static CFMetaData newCFMetaData(String ksname, String cfname)
     {
         CFMetaData metadata = CFMetaData.Builder.create(ksname, cfname)
                                                 .addPartitionKey("key", UTF8Type.instance)

test/unit/org/apache/cassandra/net/WriteCallbackInfoTest.java

@@ -0,0 +1,79 @@
+/*
+* 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.net;
+
+import java.net.InetAddress;
+import java.util.UUID;
+
+import org.junit.Test;
+
+import junit.framework.Assert;
+import org.apache.cassandra.MockSchema;
+import org.apache.cassandra.db.BufferDecoratedKey;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.Mutation;
+import org.apache.cassandra.db.PartitionColumns;
+import org.apache.cassandra.db.partitions.PartitionUpdate;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.net.MessagingService.Verb;
+import org.apache.cassandra.service.paxos.Commit;
+import org.apache.cassandra.utils.ByteBufferUtil;
+
+public class WriteCallbackInfoTest
+{
+
+    @Test
+    public void testShouldHint()
+    {
+        testShouldHint(Verb.COUNTER_MUTATION, ConsistencyLevel.ALL, true, false);
+        for (Verb verb : new Verb[] { Verb.PAXOS_COMMIT, Verb.MUTATION })
+        {
+            testShouldHint(verb, ConsistencyLevel.ALL, true, true);
+            testShouldHint(verb, ConsistencyLevel.ANY, true, false);
+            testShouldHint(verb, ConsistencyLevel.ALL, false, false);
+        }
+    }
+
+    private void testShouldHint(Verb verb, ConsistencyLevel cl, boolean allowHints, boolean expectHint)
+    {
+        Object payload = verb == Verb.PAXOS_COMMIT
+                         ? new Commit(UUID.randomUUID(), new PartitionUpdate(MockSchema.newCFMetaData("", ""), ByteBufferUtil.EMPTY_BYTE_BUFFER, PartitionColumns.NONE, 1))
+                         : new Mutation("", new BufferDecoratedKey(new Murmur3Partitioner.LongToken(0), ByteBufferUtil.EMPTY_BYTE_BUFFER));
+
+        WriteCallbackInfo wcbi = new WriteCallbackInfo(InetAddress.getLoopbackAddress(), null, new MessageOut(verb, payload, null), null, cl, allowHints);
+        Assert.assertEquals(expectHint, wcbi.shouldHint());
+        if (expectHint)
+        {
+            Assert.assertNotNull(wcbi.mutation());
+        }
+        else
+        {
+            boolean fail = false;
+            try
+            {
+                wcbi.mutation();
+            }
+            catch (Throwable t)
+            {
+                fail = true;
+            }
+            Assert.assertTrue(fail);
+        }
+    }
+}