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

@@ -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

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);
-            {
-                StorageProxy.insertBlocking(rm, consistency_level);
-            }
-            catch (TimeoutException e)
-            {
-                throw new TimedOutException();
-            }
         }
         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);

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<Row> weakReadRemote(List<ReadCommand> commands) throws IOException, UnavailableException, TimeoutException
+    private static List<Row> weakReadRemote(List<ReadCommand> 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<Row> readProtocol(List<ReadCommand> 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<Row> strongRead(List<ReadCommand> commands, int consistency_level) throws IOException, UnavailableException, TimeoutException
+    private static List<Row> strongRead(List<ReadCommand> commands, int consistency_level) throws IOException, UnavailableException, TimedOutException
     {
         List<QuorumResponseHandler<Row>> quorumResponseHandlers = new ArrayList<QuorumResponseHandler<Row>>();
         List<InetAddress[]> commandEndPoints = new ArrayList<InetAddress[]>();
@@ -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<Pair<String, ColumnFamily>> getRangeSlice(RangeSliceCommand command, int consistency_level) throws IOException, UnavailableException, TimeoutException
+    static List<Pair<String, ColumnFamily>> 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<String> getKeyRange(RangeCommand command) throws IOException, UnavailableException, TimeoutException
+    static List<String> 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);