Merge pull request #164 from hannosch/c11-exception-changes

C11 exception base class changes vs. pool retry

pycassa/pool.py

@@ -12,6 +12,7 @@
     import Queue
 
 from thrift import Thrift
+from thrift.transport.TTransport import TTransportException
 from connection import Connection
 from logging.pool_logger import PoolLogger
 from util import as_interface
@@ -128,7 +129,8 @@ def new_f(self, *args, **kwargs):
                 self._pool._decrement_overflow()
                 self._pool._clear_current()
                 raise app_exc
-            except (TimedOutException, UnavailableException, Thrift.TException,
+            except (TimedOutException, UnavailableException,
+                    TTransportException,
                     socket.error, IOError, EOFError), exc:
                 self._pool._notify_on_failure(exc, server=self.server, connection=self)
 
@@ -400,7 +402,7 @@ def _create_connection(self):
                 server = self._get_next_server()
                 wrapper = self._get_new_wrapper(server)
                 return wrapper
-            except (Thrift.TException, socket.error, IOError, EOFError), exc:
+            except (TTransportException, socket.error, IOError, EOFError), exc:
                 self._notify_on_failure(exc, server)
                 failure_count += 1
         raise AllServersUnavailable('An attempt was made to connect to each of the servers ' +

tests/test_connection_pooling.py

@@ -6,6 +6,9 @@
 from pycassa import ColumnFamily, ConnectionPool, PoolListener, InvalidRequestError,\
                     NoConnectionAvailable, MaximumRetryException, AllServersUnavailable
 from pycassa.cassandra.ttypes import ColumnPath
+from pycassa.cassandra.ttypes import InvalidRequestException
+from pycassa.cassandra.ttypes import NotFoundException
+
 
 _credentials = {'username':'jsmith', 'password':'havebadpass'}
 
@@ -486,6 +489,21 @@ def test_failure_connection_info(self):
         assert_equal(request['args'], ('greunt', ColumnPath('Counter1', None, 'col'), 1))
         assert_equal(request['kwargs'], {})
 
+    def test_pool_invalid_request(self):
+        listener = _TestListener()
+        pool = ConnectionPool(pool_size=1, max_overflow=0, recycle=10000,
+                         prefill=True, max_retries=3,
+                         keyspace='PycassaTestKeyspace',
+                         credentials=_credentials,
+                         listeners=[listener], use_threadlocal=False,
+                         server_list=['localhost:9160'])
+        cf = ColumnFamily(pool, 'Standard1')
+        # Make sure the pool doesn't hide and retries invalid requests
+        assert_raises(InvalidRequestException, cf.add, 'key', 'col')
+        assert_raises(NotFoundException, cf.get, 'none')
+        pool.dispose()
+
+
 class _TestListener(PoolListener):
 
     def __init__(self):