diff --git a/src/java/org/apache/cassandra/net/SelectionKeyHandler.java b/src/java/org/apache/cassandra/net/SelectionKeyHandler.java
index 0048b7bce7..5f9605ab24 100644
--- a/src/java/org/apache/cassandra/net/SelectionKeyHandler.java
+++ b/src/java/org/apache/cassandra/net/SelectionKeyHandler.java
@@ -65,4 +65,20 @@ public class SelectionKeyHandler
     {
         throw new UnsupportedOperationException("write() cannot be called on " + getClass().getName() + "!");
     }
+    
+    protected static void turnOnInterestOps(SelectionKey key, int ops)
+    {
+        synchronized(key)
+        {
+            key.interestOps(key.interestOps() | ops);
+        }
+    }
+    
+    protected static void turnOffInterestOps(SelectionKey key, int ops)
+    {
+        synchronized(key)
+        {
+            key.interestOps(key.interestOps() & (~ops) );
+        }
+    }
 }
diff --git a/src/java/org/apache/cassandra/net/TcpConnection.java b/src/java/org/apache/cassandra/net/TcpConnection.java
index 99c0f9f3ec..c87127e5a6 100644
--- a/src/java/org/apache/cassandra/net/TcpConnection.java
+++ b/src/java/org/apache/cassandra/net/TcpConnection.java
@@ -183,7 +183,7 @@ public class TcpConnection extends SelectionKeyHandler implements Comparable
                 if (buffer.remaining() > 0) 
                 {                   
                     pendingWrites_.add(buffer);
-                    key_.interestOps(key_.interestOps() | SelectionKey.OP_WRITE);
+                    turnOnInterestOps(key_, SelectionKey.OP_WRITE);
                 }
             }
         }
@@ -229,7 +229,7 @@ public class TcpConnection extends SelectionKeyHandler implements Comparable
                     if (buffer.remaining() > 0)
                     {
                         pendingWrites_.add(buffer);
-                        key_.interestOps(key_.interestOps() | SelectionKey.OP_WRITE);
+                        turnOnInterestOps(key_, SelectionKey.OP_WRITE);
                         condition_.await();
                     }
                 }
@@ -245,7 +245,7 @@ public class TcpConnection extends SelectionKeyHandler implements Comparable
                 */
                 if ( bytesTransferred < limit && bytesWritten != total )
                 {                    
-                    key_.interestOps(key_.interestOps() | SelectionKey.OP_WRITE);
+                    turnOnInterestOps(key_, SelectionKey.OP_WRITE);
                     condition_.await();
                 }
             }
@@ -346,17 +346,20 @@ public class TcpConnection extends SelectionKeyHandler implements Comparable
     // called in the selector thread
     public void connect(SelectionKey key)
     {       
-        key.interestOps(key.interestOps() & (~SelectionKey.OP_CONNECT));
+        turnOffInterestOps(key, SelectionKey.OP_CONNECT);
         try
         {
             if (socketChannel_.finishConnect())
             {
-                key.interestOps(key.interestOps() | SelectionKey.OP_READ);
+                turnOnInterestOps(key, SelectionKey.OP_READ);
                 
-                // this will flush the pending                
-                if (!pendingWrites_.isEmpty()) 
+                synchronized(this)
                 {
-                    key_.interestOps(key_.interestOps() | SelectionKey.OP_WRITE);
+                    // this will flush the pending                
+                    if (!pendingWrites_.isEmpty()) 
+                    {
+                        turnOnInterestOps(key_, SelectionKey.OP_WRITE);
+                    }
                 }
                 resumeStreaming();
             } 
@@ -376,7 +379,7 @@ public class TcpConnection extends SelectionKeyHandler implements Comparable
     // called in the selector thread
     public void write(SelectionKey key)
     {   
-        key.interestOps( key.interestOps() & ( ~SelectionKey.OP_WRITE ) );                
+        turnOffInterestOps(key, SelectionKey.OP_WRITE);                
         doPendingWrites();
         /*
          * This is executed only if we are in streaming mode.
@@ -415,7 +418,7 @@ public class TcpConnection extends SelectionKeyHandler implements Comparable
             {    
                 if (!pendingWrites_.isEmpty())
                 {                    
-                    key_.interestOps(key_.interestOps() | SelectionKey.OP_WRITE);
+                    turnOnInterestOps(key_, SelectionKey.OP_WRITE);
                 }
             }
         }
@@ -424,7 +427,7 @@ public class TcpConnection extends SelectionKeyHandler implements Comparable
     // called in the selector thread
     public void read(SelectionKey key)
     {
-        key.interestOps( key.interestOps() & ( ~SelectionKey.OP_READ ) );
+        turnOffInterestOps(key, SelectionKey.OP_READ);
         // publish this event onto to the TCPReadEvent Queue.
         MessagingService.getReadExecutor().execute(readWork_);
     }
@@ -486,7 +489,7 @@ public class TcpConnection extends SelectionKeyHandler implements Comparable
             }
             finally
             {
-                key_.interestOps(key_.interestOps() | SelectionKey.OP_READ);
+                turnOnInterestOps(key_, SelectionKey.OP_READ);
             }
         }
         
diff --git a/src/java/org/apache/cassandra/net/UdpConnection.java b/src/java/org/apache/cassandra/net/UdpConnection.java
index 53fddbb88b..177b73bf05 100644
--- a/src/java/org/apache/cassandra/net/UdpConnection.java
+++ b/src/java/org/apache/cassandra/net/UdpConnection.java
@@ -131,7 +131,7 @@ public class UdpConnection extends SelectionKeyHandler
     
     public void read(SelectionKey key)
     {        
-        key.interestOps( key.interestOps() & (~SelectionKey.OP_READ) );
+        turnOffInterestOps(key, SelectionKey.OP_READ);
         ByteBuffer buffer = ByteBuffer.allocate(BUFFER_SIZE);
         try
         {
@@ -160,7 +160,7 @@ public class UdpConnection extends SelectionKeyHandler
         }
         finally
         {
-            key.interestOps( key_.interestOps() | SelectionKey.OP_READ );
+            turnOnInterestOps(key_, SelectionKey.OP_READ );
         }
     }
 }
diff --git a/src/java/org/apache/cassandra/net/http/HttpConnection.java b/src/java/org/apache/cassandra/net/http/HttpConnection.java
index d0ce82846c..ce6ff4dc1f 100644
--- a/src/java/org/apache/cassandra/net/http/HttpConnection.java
+++ b/src/java/org/apache/cassandra/net/http/HttpConnection.java
@@ -136,7 +136,7 @@ public class HttpConnection extends SelectionKeyHandler implements HttpStartLine
             httpChannel_ = (SocketChannel)key.channel();
         }
         /* deregister interest for read */
-        key.interestOps( key.interestOps() & ( ~SelectionKey.OP_READ ) );
+        turnOffInterestOps(key, SelectionKey.OP_READ);
         /* Add a task to process the HTTP request */
         MessagingService.getReadExecutor().execute(httpReader_);
     }
@@ -330,7 +330,7 @@ public class HttpConnection extends SelectionKeyHandler implements HttpStartLine
         }
         finally
         {
-            httpKey_.interestOps(httpKey_.interestOps() | SelectionKey.OP_READ);
+            turnOnInterestOps(httpKey_, SelectionKey.OP_READ);
         }
     }