diff --git a/src/java/org/apache/cassandra/net/WriteCallbackInfo.java b/src/java/org/apache/cassandra/net/WriteCallbackInfo.java index 582298c5ee..bf7cc3ab0f 100644 --- a/src/java/org/apache/cassandra/net/WriteCallbackInfo.java +++ b/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; + } + } diff --git a/test/unit/org/apache/cassandra/MockSchema.java b/test/unit/org/apache/cassandra/MockSchema.java index 6b50e49abf..a406290f36 100644 --- a/test/unit/org/apache/cassandra/MockSchema.java +++ b/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) diff --git a/test/unit/org/apache/cassandra/net/WriteCallbackInfoTest.java b/test/unit/org/apache/cassandra/net/WriteCallbackInfoTest.java new file mode 100644 index 0000000000..ac726d55c1 --- /dev/null +++ b/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); + } + } +}