You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by jb...@apache.org on 2009/12/28 16:57:09 UTC
svn commit: r894192 - in /incubator/cassandra/branches/cassandra-0.5:
CHANGES.txt src/java/org/apache/cassandra/service/CassandraServer.java
src/java/org/apache/cassandra/service/StorageProxy.java
Author: jbellis
Date: Mon Dec 28 15:57:08 2009
New Revision: 894192
URL: http://svn.apache.org/viewvc?rev=894192&view=rev
Log:
expose java.util.concurrent.TimeoutException in StorageProxy methods. patch by Todd Blose; reviewed by jbellis for CASSANDRA-600
Modified:
incubator/cassandra/branches/cassandra-0.5/CHANGES.txt
incubator/cassandra/branches/cassandra-0.5/src/java/org/apache/cassandra/service/CassandraServer.java
incubator/cassandra/branches/cassandra-0.5/src/java/org/apache/cassandra/service/StorageProxy.java
Modified: incubator/cassandra/branches/cassandra-0.5/CHANGES.txt
URL: http://svn.apache.org/viewvc/incubator/cassandra/branches/cassandra-0.5/CHANGES.txt?rev=894192&r1=894191&r2=894192&view=diff
==============================================================================
--- incubator/cassandra/branches/cassandra-0.5/CHANGES.txt (original)
+++ incubator/cassandra/branches/cassandra-0.5/CHANGES.txt Mon Dec 28 15:57:08 2009
@@ -1,6 +1,8 @@
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)
0.5.0 RC1
Modified: incubator/cassandra/branches/cassandra-0.5/src/java/org/apache/cassandra/service/CassandraServer.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/branches/cassandra-0.5/src/java/org/apache/cassandra/service/CassandraServer.java?rev=894192&r1=894191&r2=894192&view=diff
==============================================================================
--- incubator/cassandra/branches/cassandra-0.5/src/java/org/apache/cassandra/service/CassandraServer.java (original)
+++ incubator/cassandra/branches/cassandra-0.5/src/java/org/apache/cassandra/service/CassandraServer.java Mon Dec 28 15:57:08 2009
@@ -22,6 +22,7 @@
import java.io.FileInputStream;
import java.io.IOException;
import java.util.*;
+import java.util.concurrent.TimeoutException;
import org.apache.log4j.Logger;
@@ -92,6 +93,10 @@
{
rows = StorageProxy.readProtocol(commands, consistency_level);
}
+ catch (TimeoutException e)
+ {
+ throw new TimedOutException();
+ }
catch (IOException e)
{
throw new RuntimeException(e);
@@ -463,7 +468,14 @@
{
if (consistency_level != ConsistencyLevel.ZERO)
{
- StorageProxy.insertBlocking(rm, consistency_level);
+ try
+ {
+ StorageProxy.insertBlocking(rm, consistency_level);
+ }
+ catch (TimeoutException e)
+ {
+ throw new TimedOutException();
+ }
}
else
{
@@ -573,6 +585,10 @@
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);
@@ -608,6 +624,10 @@
{
return StorageProxy.getKeyRange(new RangeCommand(tablename, columnFamily, startWith, stopAt, maxResults));
}
+ catch (TimeoutException e)
+ {
+ throw new TimedOutException();
+ }
catch (IOException e)
{
throw new RuntimeException(e);
Modified: incubator/cassandra/branches/cassandra-0.5/src/java/org/apache/cassandra/service/StorageProxy.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/branches/cassandra-0.5/src/java/org/apache/cassandra/service/StorageProxy.java?rev=894192&r1=894191&r2=894192&view=diff
==============================================================================
--- incubator/cassandra/branches/cassandra-0.5/src/java/org/apache/cassandra/service/StorageProxy.java (original)
+++ incubator/cassandra/branches/cassandra-0.5/src/java/org/apache/cassandra/service/StorageProxy.java Mon Dec 28 15:57:08 2009
@@ -160,7 +160,7 @@
}
}
- public static void insertBlocking(final RowMutation rm, int consistency_level) throws UnavailableException, TimedOutException
+ public static void insertBlocking(final RowMutation rm, int consistency_level) throws UnavailableException, TimeoutException
{
long startTime = System.currentTimeMillis();
try
@@ -240,10 +240,6 @@
// 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);
@@ -292,7 +288,7 @@
* @return the row associated with command.key
* @throws Exception
*/
- private static List<Row> weakReadRemote(List<ReadCommand> commands) throws IOException, UnavailableException, TimedOutException
+ private static List<Row> weakReadRemote(List<ReadCommand> commands) throws IOException, UnavailableException, TimeoutException
{
if (logger.isDebugEnabled())
logger.debug("weakreadremote reading " + StringUtils.join(commands, ", "));
@@ -314,14 +310,7 @@
for (IAsyncResult iar: iars)
{
byte[] body;
- try
- {
- body = iar.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS);
- }
- catch (TimeoutException e)
- {
- throw new TimedOutException();
- }
+ body = iar.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS);
DataInputBuffer bufIn = new DataInputBuffer();
bufIn.reset(body, body.length);
ReadResponse response = ReadResponse.serializer().deserialize(bufIn);
@@ -336,7 +325,7 @@
* 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, TimedOutException
+ throws IOException, UnavailableException, TimeoutException
{
long startTime = System.currentTimeMillis();
@@ -390,7 +379,7 @@
* 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, TimedOutException
+ private static List<Row> strongRead(List<ReadCommand> commands, int consistency_level) throws IOException, UnavailableException, TimeoutException
{
List<QuorumResponseHandler<Row>> quorumResponseHandlers = new ArrayList<QuorumResponseHandler<Row>>();
List<InetAddress[]> commandEndPoints = new ArrayList<InetAddress[]>();
@@ -445,10 +434,6 @@
if (logger.isDebugEnabled())
logger.debug("quorumResponseHandler: " + (System.currentTimeMillis() - startTime2) + " ms.");
}
- catch (TimeoutException e)
- {
- throw new TimedOutException();
- }
catch (DigestMismatchException ex)
{
if (DatabaseDescriptor.getConsistencyCheck())
@@ -466,10 +451,6 @@
if (row != null)
rows.add(row);
}
- catch (TimeoutException e)
- {
- throw new TimedOutException();
- }
catch (DigestMismatchException e)
{
// TODO should this be a thrift exception?
@@ -512,7 +493,7 @@
return rows;
}
- static List<Pair<String, ColumnFamily>> getRangeSlice(RangeSliceCommand command, int consistency_level) throws IOException, UnavailableException, TimedOutException
+ static List<Pair<String, ColumnFamily>> getRangeSlice(RangeSliceCommand command, int consistency_level) throws IOException, UnavailableException, TimeoutException
{
long startTime = System.currentTimeMillis();
TokenMetadata tokenMetadata = StorageService.instance().getTokenMetadata();
@@ -562,10 +543,6 @@
{
rows.putAll(handler.get());
}
- catch (TimeoutException e)
- {
- throw new TimedOutException();
- }
catch (DigestMismatchException e)
{
throw new AssertionError(e); // no digests in range slices yet
@@ -594,7 +571,7 @@
return results;
}
- static List<String> getKeyRange(RangeCommand command) throws IOException, UnavailableException, TimedOutException
+ static List<String> getKeyRange(RangeCommand command) throws IOException, UnavailableException, TimeoutException
{
long startTime = System.currentTimeMillis();
TokenMetadata tokenMetadata = StorageService.instance().getTokenMetadata();
@@ -612,14 +589,8 @@
// read response
byte[] responseBody;
- try
- {
- responseBody = iar.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS);
- }
- catch (TimeoutException e)
- {
- throw new TimedOutException();
- }
+ responseBody = iar.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS);
+
RangeReply rangeReply = RangeReply.read(responseBody);
uniqueKeys.addAll(rangeReply.keys);