natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix handling of RejectedExecution in sync Thrift server 

patch by Christian Rolf and jbellis for CASSANDRA-6788
This commit is contained in:
Jonathan Ellis 2014-03-17 10:50:08 -05:00
parent 1e64972be5
commit 56e2b4ac02
2 changed files with 14 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
2.0.7
* Fix handling of RejectedExecution in sync Thrift server (CASSANDRA-6788)
* Log more information when exceeding tombstone_warn_threshold (CASSANDRA-6865)
* Fix truncate to not abort due to unreachable fat clients (CASSANDRA-6864)
* Fix schema concurrency exceptions (CASSANDRA-6841)

25
src/java/org/apache/cassandra/thrift/CustomTThreadPoolServer.java
View File

@ -21,6 +21,7 @@ import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.SocketTimeoutException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@ -97,7 +98,7 @@ public class CustomTThreadPoolServer extends TServer
// block until we are under max clients
while (activeClients.get() >= args.maxWorkerThreads)
{
Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
Uninterruptibles.sleepUninterruptibly(10, TimeUnit.MILLISECONDS);
}
try
@ -117,6 +118,12 @@ public class CustomTThreadPoolServer extends TServer
logger.warn("Transport error occurred during acceptance of message.", ttx);
}
}
catch (RejectedExecutionException e)
{
// worker thread decremented activeClients but hadn't finished exiting
logger.debug("Dropping client connection because our limit of {} has been reached", args.maxWorkerThreads);
continue;
}
if (activeClients.get() >= args.maxWorkerThreads)
logger.warn("Maximum number of clients " + args.maxWorkerThreads + " reached");
@ -213,19 +220,13 @@ public class CustomTThreadPoolServer extends TServer
}
finally
{
activeClients.decrementAndGet();
if (socket != null)
ThriftSessionManager.instance.connectionComplete(socket);
}
if (inputTransport != null)
{
inputTransport.close();
}
if (outputTransport != null)
{
outputTransport.close();
if (inputTransport != null)
inputTransport.close();
if (outputTransport != null)
outputTransport.close();
activeClients.decrementAndGet();
}
}
}