Merge branch 'cassandra-2.1' into trunk

Conflicts:
	src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java

CHANGES.txt

@@ -84,6 +84,7 @@ Merged from 1.2:
  * Optimize Cell liveness checks and clean up Cell (CASSANDRA-7119)
  * Support consistent range movements (CASSANDRA-2434)
 Merged from 2.0:
+ * Starting threads in OutboundTcpConnectionPool constructor causes race conditions (CASSANDRA-7177)
  * Allow overriding cassandra-rackdc.properties file (CASSANDRA-7072)
  * Set JMX RMI port to 7199 (CASSANDRA-7087)
  * Use LOCAL_QUORUM for data reads at LOCAL_SERIAL (CASSANDRA-6939)

src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java

@@ -713,6 +713,9 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF
                 if (stmt.isCounter())
                     throw new InvalidRequestException("Conditional updates are not supported on counter tables");
 
+                if (attrs.timestamp != null)
+                    throw new InvalidRequestException("Cannot provide custom timestamp for conditional update");
+
                 if (ifNotExists)
                 {
                     // To have both 'IF NOT EXISTS' and some other conditions doesn't make sense.

src/java/org/apache/cassandra/net/MessagingService.java

@@ -512,11 +512,11 @@ public final class MessagingService implements MessagingServiceMBean
             cp = new OutboundTcpConnectionPool(to);
             OutboundTcpConnectionPool existingPool = connectionManagers.putIfAbsent(to, cp);
             if (existingPool != null)
-            {
-                cp.close();
                 cp = existingPool;
-            }
+            else
+                cp.start();
         }
+        cp.waitForStarted();
         return cp;
     }
     

src/java/org/apache/cassandra/net/OutboundTcpConnectionPool.java

@@ -22,6 +22,8 @@ import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.nio.channels.SocketChannel;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.cassandra.concurrent.Stage;
 import org.apache.cassandra.config.Config;
@@ -36,6 +38,7 @@ public class OutboundTcpConnectionPool
 {
     // pointer for the real Address.
     private final InetAddress id;
+    private final CountDownLatch started;
     public final OutboundTcpConnection cmdCon;
     public final OutboundTcpConnection ackCon;
     // pointer to the reseted Address.
@@ -46,13 +49,10 @@ public class OutboundTcpConnectionPool
     {
         id = remoteEp;
         resetedEndpoint = SystemKeyspace.getPreferredIP(remoteEp);
+        started = new CountDownLatch(1);
 
         cmdCon = new OutboundTcpConnection(this);
-        cmdCon.start();
         ackCon = new OutboundTcpConnection(this);
-        ackCon.start();
-
-        metrics = new ConnectionMetrics(id, this);
     }
 
     /**
@@ -167,14 +167,45 @@ public class OutboundTcpConnectionPool
         }
         return true;
     }
+    
+    public void start()
+    {
+        cmdCon.start();
+        ackCon.start();
 
-   public void close()
+        metrics = new ConnectionMetrics(id, this);
+        
+        started.countDown();
+    }
+    
+    public void waitForStarted()
+    {
+        if (started.getCount() == 0)
+            return;
+
+        boolean error = false;
+        try
+        {
+            if (!started.await(1, TimeUnit.MINUTES))
+                error = true;
+        }
+        catch (InterruptedException e)
+        {
+            Thread.currentThread().interrupt();
+            error = true;
+        }
+        if (error)
+            throw new IllegalStateException(String.format("Connections to %s are not started!", id.getHostAddress()));
+    }
+
+    public void close()
     {
         // these null guards are simply for tests
         if (ackCon != null)
             ackCon.closeSocket(true);
         if (cmdCon != null)
             cmdCon.closeSocket(true);
+        
         metrics.release();
     }
 }