From 44d5172c4eaa05ff2651a018ebeb1ae0e77d98e6 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Mon, 4 Jan 2010 17:30:28 +0000 Subject: [PATCH] revert 894192, leaving CASSANDRA-600 only in trunk. patch by jbellis git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/branches/cassandra-0.5@895716 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 4 +- .../cassandra/service/CassandraServer.java | 22 +-------- .../cassandra/service/StorageProxy.java | 47 +++++++++++++++---- 3 files changed, 40 insertions(+), 33 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index a634934d25..624e94f399 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,11 +1,9 @@ 0.5.0 RC2 * fix bugs in converting get_slice_range results to Thrift (CASSANDRA-647, CASSANDRA-649) - * expose java.util.concurrent.TimeoutException in StorageProxy methods - (CASSANDRA-600) * TcpConnectionManager was holding on to disconnected connections, giving the false indication they were being used. (CASSANDRA-651) - * Remove duplicated write. (CASSANDRA-662) + * Avoid redundant write of the same mutation. (CASSANDRA-662) 0.5.0 RC1 diff --git a/src/java/org/apache/cassandra/service/CassandraServer.java b/src/java/org/apache/cassandra/service/CassandraServer.java index edc13ecc30..77bde86cd7 100644 --- a/src/java/org/apache/cassandra/service/CassandraServer.java +++ b/src/java/org/apache/cassandra/service/CassandraServer.java @@ -22,7 +22,6 @@ import java.io.BufferedInputStream; import java.io.FileInputStream; import java.io.IOException; import java.util.*; -import java.util.concurrent.TimeoutException; import org.apache.log4j.Logger; @@ -93,10 +92,6 @@ public class CassandraServer implements Cassandra.Iface { rows = StorageProxy.readProtocol(commands, consistency_level); } - catch (TimeoutException e) - { - throw new TimedOutException(); - } catch (IOException e) { throw new RuntimeException(e); @@ -468,14 +463,7 @@ public class CassandraServer implements Cassandra.Iface { if (consistency_level != ConsistencyLevel.ZERO) { - try - { - StorageProxy.insertBlocking(rm, consistency_level); - } - catch (TimeoutException e) - { - throw new TimedOutException(); - } + StorageProxy.insertBlocking(rm, consistency_level); } else { @@ -585,10 +573,6 @@ public class CassandraServer implements Cassandra.Iface rows = StorageProxy.getRangeSlice(new RangeSliceCommand(keyspace, column_parent, predicate, startKey, finishKey, maxRows), consistency_level); assert rows != null; } - catch (TimeoutException e) - { - throw new TimedOutException(); - } catch (IOException e) { throw new RuntimeException(e); @@ -624,10 +608,6 @@ public class CassandraServer implements Cassandra.Iface { return StorageProxy.getKeyRange(new RangeCommand(tablename, columnFamily, startWith, stopAt, maxResults)); } - catch (TimeoutException e) - { - throw new TimedOutException(); - } catch (IOException e) { throw new RuntimeException(e); diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 100e1a18a2..79dfb72008 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -160,7 +160,7 @@ public class StorageProxy implements StorageProxyMBean } } - public static void insertBlocking(final RowMutation rm, int consistency_level) throws UnavailableException, TimeoutException + public static void insertBlocking(final RowMutation rm, int consistency_level) throws UnavailableException, TimedOutException { long startTime = System.currentTimeMillis(); try @@ -240,6 +240,10 @@ public class StorageProxy implements StorageProxyMBean // wait for writes. throws timeoutexception if necessary responseHandler.get(); } + catch (TimeoutException e) + { + throw new TimedOutException(); + } catch (IOException e) { throw new RuntimeException("error writing key " + rm.key(), e); @@ -288,7 +292,7 @@ public class StorageProxy implements StorageProxyMBean * @return the row associated with command.key * @throws Exception */ - private static List weakReadRemote(List commands) throws IOException, UnavailableException, TimeoutException + private static List weakReadRemote(List commands) throws IOException, UnavailableException, TimedOutException { if (logger.isDebugEnabled()) logger.debug("weakreadremote reading " + StringUtils.join(commands, ", ")); @@ -310,7 +314,14 @@ public class StorageProxy implements StorageProxyMBean for (IAsyncResult iar: iars) { byte[] body; - body = iar.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS); + try + { + body = iar.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS); + } + catch (TimeoutException e) + { + throw new TimedOutException(); + } DataInputBuffer bufIn = new DataInputBuffer(); bufIn.reset(body, body.length); ReadResponse response = ReadResponse.serializer().deserialize(bufIn); @@ -325,7 +336,7 @@ public class StorageProxy implements StorageProxyMBean * a specific set of column names from a given column family. */ public static List readProtocol(List commands, int consistency_level) - throws IOException, UnavailableException, TimeoutException + throws IOException, UnavailableException, TimedOutException { long startTime = System.currentTimeMillis(); @@ -379,7 +390,7 @@ public class StorageProxy implements StorageProxyMBean * 7. else carry out read repair by getting data from all the nodes. // 5. return success */ - private static List strongRead(List commands, int consistency_level) throws IOException, UnavailableException, TimeoutException + private static List strongRead(List commands, int consistency_level) throws IOException, UnavailableException, TimedOutException { List> quorumResponseHandlers = new ArrayList>(); List commandEndPoints = new ArrayList(); @@ -434,6 +445,10 @@ public class StorageProxy implements StorageProxyMBean if (logger.isDebugEnabled()) logger.debug("quorumResponseHandler: " + (System.currentTimeMillis() - startTime2) + " ms."); } + catch (TimeoutException e) + { + throw new TimedOutException(); + } catch (DigestMismatchException ex) { if (DatabaseDescriptor.getConsistencyCheck()) @@ -451,6 +466,10 @@ public class StorageProxy implements StorageProxyMBean if (row != null) rows.add(row); } + catch (TimeoutException e) + { + throw new TimedOutException(); + } catch (DigestMismatchException e) { // TODO should this be a thrift exception? @@ -493,7 +512,7 @@ public class StorageProxy implements StorageProxyMBean return rows; } - static List> getRangeSlice(RangeSliceCommand command, int consistency_level) throws IOException, UnavailableException, TimeoutException + static List> getRangeSlice(RangeSliceCommand command, int consistency_level) throws IOException, UnavailableException, TimedOutException { long startTime = System.currentTimeMillis(); TokenMetadata tokenMetadata = StorageService.instance().getTokenMetadata(); @@ -543,6 +562,10 @@ public class StorageProxy implements StorageProxyMBean { rows.putAll(handler.get()); } + catch (TimeoutException e) + { + throw new TimedOutException(); + } catch (DigestMismatchException e) { throw new AssertionError(e); // no digests in range slices yet @@ -571,7 +594,7 @@ public class StorageProxy implements StorageProxyMBean return results; } - static List getKeyRange(RangeCommand command) throws IOException, UnavailableException, TimeoutException + static List getKeyRange(RangeCommand command) throws IOException, UnavailableException, TimedOutException { long startTime = System.currentTimeMillis(); TokenMetadata tokenMetadata = StorageService.instance().getTokenMetadata(); @@ -589,8 +612,14 @@ public class StorageProxy implements StorageProxyMBean // read response byte[] responseBody; - responseBody = iar.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS); - + try + { + responseBody = iar.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS); + } + catch (TimeoutException e) + { + throw new TimedOutException(); + } RangeReply rangeReply = RangeReply.read(responseBody); uniqueKeys.addAll(rangeReply.keys);