r/m unnecessary IWriteResponseHandler (all implementations inherit AbstractWriteResponseHandler)
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/84c06577 Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/84c06577 Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/84c06577 Branch: refs/heads/trunk Commit: 84c06577dcfdfb227c2440d6100515e6f05d698c Parents: f92fb22 Author: Jonathan Ellis <[email protected]> Authored: Thu Sep 20 16:56:49 2012 -0500 Committer: Jonathan Ellis <[email protected]> Committed: Fri Sep 21 10:06:56 2012 -0500 ---------------------------------------------------------------------- .../org/apache/cassandra/db/CounterColumn.java | 4 +- .../apache/cassandra/db/HintedHandOffManager.java | 2 +- .../locator/AbstractReplicationStrategy.java | 4 +- .../service/AbstractWriteResponseHandler.java | 3 +- .../DatacenterSyncWriteResponseHandler.java | 2 +- .../service/DatacenterWriteResponseHandler.java | 2 +- .../cassandra/service/IWriteResponseHandler.java | 28 --------- .../org/apache/cassandra/service/StorageProxy.java | 46 +++++++------- .../cassandra/service/WriteResponseHandler.java | 4 +- 9 files changed, 34 insertions(+), 61 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/84c06577/src/java/org/apache/cassandra/db/CounterColumn.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/CounterColumn.java b/src/java/org/apache/cassandra/db/CounterColumn.java index c501bf1..33f391e 100644 --- a/src/java/org/apache/cassandra/db/CounterColumn.java +++ b/src/java/org/apache/cassandra/db/CounterColumn.java @@ -37,7 +37,7 @@ import org.apache.cassandra.exceptions.RequestExecutionException; import org.apache.cassandra.io.IColumnSerializer; import org.apache.cassandra.io.util.DataOutputBuffer; import org.apache.cassandra.utils.Allocator; -import org.apache.cassandra.service.IWriteResponseHandler; +import org.apache.cassandra.service.AbstractWriteResponseHandler; import org.apache.cassandra.service.StorageProxy; import org.apache.cassandra.utils.*; @@ -366,7 +366,7 @@ public class CounterColumn extends Column StorageProxy.performWrite(rm, ConsistencyLevel.ANY, localDataCenter, new StorageProxy.WritePerformer() { - public void apply(IMutation mutation, Collection<InetAddress> targets, IWriteResponseHandler responseHandler, String localDataCenter, ConsistencyLevel consistency_level) + public void apply(IMutation mutation, Collection<InetAddress> targets, AbstractWriteResponseHandler responseHandler, String localDataCenter, ConsistencyLevel consistency_level) throws IOException, OverloadedException { // We should only send to the remote replica, not the local one http://git-wip-us.apache.org/repos/asf/cassandra/blob/84c06577/src/java/org/apache/cassandra/db/HintedHandOffManager.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/HintedHandOffManager.java b/src/java/org/apache/cassandra/db/HintedHandOffManager.java index 064033b..6c2f951 100644 --- a/src/java/org/apache/cassandra/db/HintedHandOffManager.java +++ b/src/java/org/apache/cassandra/db/HintedHandOffManager.java @@ -126,7 +126,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean private static void sendMutation(InetAddress endpoint, MessageOut<?> message) throws WriteTimeoutException { - IWriteResponseHandler responseHandler = WriteResponseHandler.create(endpoint); + AbstractWriteResponseHandler responseHandler = WriteResponseHandler.create(endpoint); MessagingService.instance().sendRR(message, endpoint, responseHandler); responseHandler.get(); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/84c06577/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java index 0ba7dcc..2a68207 100644 --- a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java +++ b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java @@ -32,7 +32,7 @@ import org.apache.cassandra.dht.RingPosition; import org.apache.cassandra.dht.Token; import org.apache.cassandra.service.DatacenterSyncWriteResponseHandler; import org.apache.cassandra.service.DatacenterWriteResponseHandler; -import org.apache.cassandra.service.IWriteResponseHandler; +import org.apache.cassandra.service.AbstractWriteResponseHandler; import org.apache.cassandra.service.WriteResponseHandler; import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.utils.FBUtilities; @@ -114,7 +114,7 @@ public abstract class AbstractReplicationStrategy */ public abstract List<InetAddress> calculateNaturalEndpoints(Token searchToken, TokenMetadata tokenMetadata); - public IWriteResponseHandler getWriteResponseHandler(Collection<InetAddress> writeEndpoints, ConsistencyLevel consistency_level, Runnable callback) + public AbstractWriteResponseHandler getWriteResponseHandler(Collection<InetAddress> writeEndpoints, ConsistencyLevel consistency_level, Runnable callback) { if (consistency_level == ConsistencyLevel.LOCAL_QUORUM) { http://git-wip-us.apache.org/repos/asf/cassandra/blob/84c06577/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java b/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java index 4b3244e..3fc5b7b 100644 --- a/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java +++ b/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java @@ -24,11 +24,12 @@ import java.util.concurrent.TimeUnit; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.exceptions.UnavailableException; import org.apache.cassandra.exceptions.WriteTimeoutException; +import org.apache.cassandra.net.IAsyncCallback; import org.apache.cassandra.net.MessageIn; import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.utils.SimpleCondition; -public abstract class AbstractWriteResponseHandler implements IWriteResponseHandler +public abstract class AbstractWriteResponseHandler implements IAsyncCallback { private final SimpleCondition condition = new SimpleCondition(); protected final long startTime; http://git-wip-us.apache.org/repos/asf/cassandra/blob/84c06577/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java b/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java index 85afade..08b9c48 100644 --- a/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java +++ b/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java @@ -66,7 +66,7 @@ public class DatacenterSyncWriteResponseHandler extends AbstractWriteResponseHan } } - public static IWriteResponseHandler create(Collection<InetAddress> writeEndpoints, ConsistencyLevel consistencyLevel, String table, Runnable callback) + public static AbstractWriteResponseHandler create(Collection<InetAddress> writeEndpoints, ConsistencyLevel consistencyLevel, String table, Runnable callback) { return new DatacenterSyncWriteResponseHandler(writeEndpoints, consistencyLevel, table, callback); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/84c06577/src/java/org/apache/cassandra/service/DatacenterWriteResponseHandler.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/DatacenterWriteResponseHandler.java b/src/java/org/apache/cassandra/service/DatacenterWriteResponseHandler.java index 2821e3b..ad984a1 100644 --- a/src/java/org/apache/cassandra/service/DatacenterWriteResponseHandler.java +++ b/src/java/org/apache/cassandra/service/DatacenterWriteResponseHandler.java @@ -49,7 +49,7 @@ public class DatacenterWriteResponseHandler extends WriteResponseHandler assert consistencyLevel == ConsistencyLevel.LOCAL_QUORUM; } - public static IWriteResponseHandler create(Collection<InetAddress> writeEndpoints, ConsistencyLevel consistencyLevel, String table, Runnable callback) + public static AbstractWriteResponseHandler create(Collection<InetAddress> writeEndpoints, ConsistencyLevel consistencyLevel, String table, Runnable callback) { return new DatacenterWriteResponseHandler(writeEndpoints, consistencyLevel, table, callback); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/84c06577/src/java/org/apache/cassandra/service/IWriteResponseHandler.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/IWriteResponseHandler.java b/src/java/org/apache/cassandra/service/IWriteResponseHandler.java deleted file mode 100644 index aa77635..0000000 --- a/src/java/org/apache/cassandra/service/IWriteResponseHandler.java +++ /dev/null @@ -1,28 +0,0 @@ -/* - * 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.service; - -import org.apache.cassandra.net.IAsyncCallback; -import org.apache.cassandra.exceptions.WriteTimeoutException; -import org.apache.cassandra.exceptions.UnavailableException; - -public interface IWriteResponseHandler extends IAsyncCallback -{ - public void get() throws WriteTimeoutException; - public void assureSufficientLiveNodes() throws UnavailableException; -} http://git-wip-us.apache.org/repos/asf/cassandra/blob/84c06577/src/java/org/apache/cassandra/service/StorageProxy.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 08aad56..8f048ee 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -107,7 +107,7 @@ public class StorageProxy implements StorageProxyMBean { public void apply(IMutation mutation, Collection<InetAddress> targets, - IWriteResponseHandler responseHandler, + AbstractWriteResponseHandler responseHandler, String localDataCenter, ConsistencyLevel consistency_level) throws IOException, OverloadedException @@ -127,7 +127,7 @@ public class StorageProxy implements StorageProxyMBean { public void apply(IMutation mutation, Collection<InetAddress> targets, - IWriteResponseHandler responseHandler, + AbstractWriteResponseHandler responseHandler, String localDataCenter, ConsistencyLevel consistency_level) throws IOException @@ -144,7 +144,7 @@ public class StorageProxy implements StorageProxyMBean { public void apply(IMutation mutation, Collection<InetAddress> targets, - IWriteResponseHandler responseHandler, + AbstractWriteResponseHandler responseHandler, String localDataCenter, ConsistencyLevel consistency_level) throws IOException @@ -174,7 +174,7 @@ public class StorageProxy implements StorageProxyMBean final String localDataCenter = DatabaseDescriptor.getEndpointSnitch().getDatacenter(FBUtilities.getBroadcastAddress()); long startTime = System.nanoTime(); - List<IWriteResponseHandler> responseHandlers = new ArrayList<IWriteResponseHandler>(mutations.size()); + List<AbstractWriteResponseHandler> responseHandlers = new ArrayList<AbstractWriteResponseHandler>(mutations.size()); IMutation mostRecentMutation = null; try @@ -193,7 +193,7 @@ public class StorageProxy implements StorageProxyMBean } // wait for writes. throws TimeoutException if necessary - for (IWriteResponseHandler responseHandler : responseHandlers) + for (AbstractWriteResponseHandler responseHandler : responseHandlers) { responseHandler.get(); } @@ -301,7 +301,7 @@ public class StorageProxy implements StorageProxyMBean throws WriteTimeoutException { RowMutation rm = BatchlogManager.getBatchlogMutationFor(mutations, uuid); - IWriteResponseHandler handler = WriteResponseHandler.create(endpoints, ConsistencyLevel.ONE, Table.SYSTEM_KS, null); + AbstractWriteResponseHandler handler = WriteResponseHandler.create(endpoints, ConsistencyLevel.ONE, Table.SYSTEM_KS, null); try { @@ -326,7 +326,7 @@ public class StorageProxy implements StorageProxyMBean { RowMutation rm = new RowMutation(Table.SYSTEM_KS, UUIDType.instance.decompose(uuid)); rm.delete(new QueryPath(SystemTable.BATCHLOG_CF), FBUtilities.timestampMicros()); - IWriteResponseHandler handler = WriteResponseHandler.create(endpoints, ConsistencyLevel.ANY, Table.SYSTEM_KS, null); + AbstractWriteResponseHandler handler = WriteResponseHandler.create(endpoints, ConsistencyLevel.ANY, Table.SYSTEM_KS, null); try { @@ -389,7 +389,7 @@ public class StorageProxy implements StorageProxyMBean * @param callback an optional callback to be run if and when the write is * successful. */ - public static IWriteResponseHandler performWrite(IMutation mutation, + public static AbstractWriteResponseHandler performWrite(IMutation mutation, ConsistencyLevel consistency_level, String localDataCenter, WritePerformer performer, @@ -401,7 +401,7 @@ public class StorageProxy implements StorageProxyMBean Collection<InetAddress> writeEndpoints = getWriteEndpoints(table, mutation.key()); - IWriteResponseHandler responseHandler = rs.getWriteResponseHandler(writeEndpoints, consistency_level, callback); + AbstractWriteResponseHandler responseHandler = rs.getWriteResponseHandler(writeEndpoints, consistency_level, callback); // exit early if we can't fulfill the CL at this time responseHandler.assureSufficientLiveNodes(); @@ -415,18 +415,18 @@ public class StorageProxy implements StorageProxyMBean { AbstractReplicationStrategy rs = Table.open(mutation.getTable()).getReplicationStrategy(); Collection<InetAddress> writeEndpoints = getWriteEndpoints(mutation.getTable(), mutation.key()); - IWriteResponseHandler responseHandler = rs.getWriteResponseHandler(writeEndpoints, consistency_level, null); + AbstractWriteResponseHandler responseHandler = rs.getWriteResponseHandler(writeEndpoints, consistency_level, null); return new WriteResponseHandlerWrapper(responseHandler, mutation, writeEndpoints); } // used by atomic_batch_mutate to decouple availability check from the write itself, caches consistency level and endpoints. private static class WriteResponseHandlerWrapper { - final IWriteResponseHandler handler; + final AbstractWriteResponseHandler handler; final RowMutation mutation; final Collection<InetAddress> endpoints; - WriteResponseHandlerWrapper(IWriteResponseHandler handler, RowMutation mutation, Collection<InetAddress> endpoints) + WriteResponseHandlerWrapper(AbstractWriteResponseHandler handler, RowMutation mutation, Collection<InetAddress> endpoints) { this.handler = handler; this.mutation = mutation; @@ -499,7 +499,7 @@ public class StorageProxy implements StorageProxyMBean */ public static void sendToHintedEndpoints(final RowMutation rm, Collection<InetAddress> targets, - IWriteResponseHandler responseHandler, + AbstractWriteResponseHandler responseHandler, String localDataCenter, ConsistencyLevel consistency_level) throws IOException, OverloadedException @@ -558,7 +558,7 @@ public class StorageProxy implements StorageProxyMBean public static Future<Void> scheduleLocalHint(final RowMutation mutation, final InetAddress target, - final IWriteResponseHandler responseHandler, + final AbstractWriteResponseHandler responseHandler, final ConsistencyLevel consistencyLevel) { // Hint of itself doesn't make sense. @@ -610,7 +610,7 @@ public class StorageProxy implements StorageProxyMBean /** * for each datacenter, send a message to one node to relay the write to other replicas */ - private static void sendMessages(String localDataCenter, Map<String, Multimap<MessageOut, InetAddress>> dcMessages, IWriteResponseHandler handler) + private static void sendMessages(String localDataCenter, Map<String, Multimap<MessageOut, InetAddress>> dcMessages, AbstractWriteResponseHandler handler) throws IOException { for (Map.Entry<String, Multimap<MessageOut, InetAddress>> entry: dcMessages.entrySet()) @@ -628,7 +628,7 @@ public class StorageProxy implements StorageProxyMBean } } - private static void sendMessagesToOneDC(MessageOut message, Collection<InetAddress> targets, boolean localDC, IWriteResponseHandler handler) throws IOException + private static void sendMessagesToOneDC(MessageOut message, Collection<InetAddress> targets, boolean localDC, AbstractWriteResponseHandler handler) throws IOException { Iterator<InetAddress> iter = targets.iterator(); InetAddress target = iter.next(); @@ -667,7 +667,7 @@ public class StorageProxy implements StorageProxyMBean logger.debug("Sending message to: " + target + " with ID " + id); } - private static void insertLocal(final RowMutation rm, final IWriteResponseHandler responseHandler) + private static void insertLocal(final RowMutation rm, final AbstractWriteResponseHandler responseHandler) { if (logger.isDebugEnabled()) logger.debug("insert writing local " + rm.toString(true)); @@ -696,7 +696,7 @@ public class StorageProxy implements StorageProxyMBean * quicker response and because the WriteResponseHandlers don't make it easy to send back an error. We also always gather * the write latencies at the coordinator node to make gathering point similar to the case of standard writes. */ - public static IWriteResponseHandler mutateCounter(CounterMutation cm, String localDataCenter) throws UnavailableException, OverloadedException, IOException + public static AbstractWriteResponseHandler mutateCounter(CounterMutation cm, String localDataCenter) throws UnavailableException, OverloadedException, IOException { InetAddress endpoint = findSuitableEndpoint(cm.getTable(), cm.key(), localDataCenter, cm.consistency()); @@ -714,7 +714,7 @@ public class StorageProxy implements StorageProxyMBean rs.getWriteResponseHandler(writeEndpoints, cm.consistency(), null).assureSufficientLiveNodes(); // Forward the actual update to the chosen leader replica - IWriteResponseHandler responseHandler = WriteResponseHandler.create(endpoint); + AbstractWriteResponseHandler responseHandler = WriteResponseHandler.create(endpoint); if (logger.isDebugEnabled()) logger.debug("forwarding counter update of key " + ByteBufferUtil.bytesToHex(cm.key()) + " to " + endpoint); @@ -761,7 +761,7 @@ public class StorageProxy implements StorageProxyMBean // Must be called on a replica of the mutation. This replica becomes the // leader of this mutation. - public static IWriteResponseHandler applyCounterMutationOnLeader(CounterMutation cm, String localDataCenter, Runnable callback) + public static AbstractWriteResponseHandler applyCounterMutationOnLeader(CounterMutation cm, String localDataCenter, Runnable callback) throws UnavailableException, IOException, OverloadedException { return performWrite(cm, cm.consistency(), localDataCenter, counterWritePerformer, callback); @@ -769,7 +769,7 @@ public class StorageProxy implements StorageProxyMBean // Same as applyCounterMutationOnLeader but must with the difference that it use the MUTATION stage to execute the write (while // applyCounterMutationOnLeader assumes it is on the MUTATION stage already) - public static IWriteResponseHandler applyCounterMutationOnCoordinator(CounterMutation cm, String localDataCenter) + public static AbstractWriteResponseHandler applyCounterMutationOnCoordinator(CounterMutation cm, String localDataCenter) throws UnavailableException, IOException, OverloadedException { return performWrite(cm, cm.consistency(), localDataCenter, counterWriteOnCoordinatorPerformer, null); @@ -777,7 +777,7 @@ public class StorageProxy implements StorageProxyMBean private static Runnable counterWriteTask(final IMutation mutation, final Collection<InetAddress> targets, - final IWriteResponseHandler responseHandler, + final AbstractWriteResponseHandler responseHandler, final String localDataCenter, final ConsistencyLevel consistency_level) { @@ -1478,7 +1478,7 @@ public class StorageProxy implements StorageProxyMBean public interface WritePerformer { - public void apply(IMutation mutation, Collection<InetAddress> targets, IWriteResponseHandler responseHandler, String localDataCenter, ConsistencyLevel consistency_level) throws IOException, OverloadedException; + public void apply(IMutation mutation, Collection<InetAddress> targets, AbstractWriteResponseHandler responseHandler, String localDataCenter, ConsistencyLevel consistency_level) throws IOException, OverloadedException; } private static abstract class DroppableRunnable implements Runnable http://git-wip-us.apache.org/repos/asf/cassandra/blob/84c06577/src/java/org/apache/cassandra/service/WriteResponseHandler.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/WriteResponseHandler.java b/src/java/org/apache/cassandra/service/WriteResponseHandler.java index cc803cb..0d9e116 100644 --- a/src/java/org/apache/cassandra/service/WriteResponseHandler.java +++ b/src/java/org/apache/cassandra/service/WriteResponseHandler.java @@ -56,12 +56,12 @@ public class WriteResponseHandler extends AbstractWriteResponseHandler responses = new AtomicInteger(1); } - public static IWriteResponseHandler create(Collection<InetAddress> writeEndpoints, ConsistencyLevel consistencyLevel, String table, Runnable callback) + public static AbstractWriteResponseHandler create(Collection<InetAddress> writeEndpoints, ConsistencyLevel consistencyLevel, String table, Runnable callback) { return new WriteResponseHandler(writeEndpoints, consistencyLevel, table, callback); } - public static IWriteResponseHandler create(InetAddress endpoint) + public static AbstractWriteResponseHandler create(InetAddress endpoint) { return new WriteResponseHandler(endpoint); }
