merge from 1.0

CHANGES.txt

@@ -64,6 +64,13 @@
  * Don't ignore IOException during compaction (CASSANDRA-3655)
  * Fix assertion error for CF with gc_grace=0 (CASSANDRA-3579)
  * Shutdown ParallelCompaction reducer executor after use (CASSANDRA-3711)
+ * Avoid < 0 value for pending tasks in leveled compaction (CASSANDRA-3693)
+ * Support TimeUUID in CassandraStorage (CASSANDRA-3327)
+ * Check schema is ready before continuin boostrapping (CASSANDRA-3629)
+ * Catch overflows during parsing of chunk_length_kb (CASSANDRA-3644)
+ * Improve stream protocol mismatch errors (CASSANDRA-3652)
+ * Avoid multiple thread doing HH to the same target (CASSANDRA-3681)
+ * Add JMX property for rp_timeout_in_ms (CASSANDRA-2940)
 Merged from 0.8:
  * avoid logging (harmless) exception when GC takes < 1ms (CASSANDRA-3656)
  * prevent new nodes from thinking down nodes are up forever (CASSANDRA-3626)

NEWS.txt

@@ -67,6 +67,14 @@ Features
     - Streaming is now multithreaded.
 
 
+1.0.7
+=====
+
+Upgrading
+---------
+    - Nothing specific to 1.0.7, please report to instruction for 1.0.6
+
+
 1.0.6
 =====
 
@@ -80,6 +88,13 @@ Upgrading
       setting the right value and then run scrub on the column family.
     - Please report to instruction for 1.0.5 if coming from an older version.
 
+Other
+-----
+    - Adds new setstreamthroughput to nodetool to configure streaming
+      throttling
+    - Adds JMX property to get/set rpc_timeout_in_ms at runtime
+    - Allow configuring (per-CF) bloom_filter_fp_chance
+
 
 1.0.5
 =====

bin/cqlsh

@@ -1021,6 +1021,33 @@ class Shell(cmd.Cmd):
         Cassandra documentation.
         """
 
+    def help_insert(self):
+        print """
+        INSERT INTO <columnFamily>
+                    ( <keyname>, <colname1> [, <colname2> [, ...]] )
+               VALUES ( <keyval>, <colval1> [, <colval2> [, ...]] )
+               [USING CONSISTENCY <consistencylevel>
+                 [AND TIMESTAMP <timestamp>]
+                 [AND TTL <timeToLive]];
+
+        An INSERT is used to write one or more columns to a record in a
+        Cassandra column family. No results are returned.
+
+        The first column name in the INSERT list must be the name of the
+        column family key. Also, there must be more than one column name
+        specified (Cassandra rows are not considered to exist with only
+        a key and no associated columns).
+
+        Unlike in SQL, the semantics of INSERT and UPDATE are identical.
+        In either case a record is created if none existed before, and
+        udpated when it does. For more information, see one of the
+        following:
+
+          HELP UPDATE
+          HELP UPDATE_USING
+          HELP CONSISTENCYLEVEL
+        """
+
     def help_update(self):
         print """
         UPDATE <columnFamily> [USING CONSISTENCY <consistencylevel>

debian/changelog

@@ -1,3 +1,9 @@
+cassandra (1.0.7) unstable; urgency=low
+
+  * New release
+
+ -- Sylvain Lebresne <slebresne@apache.org>  Wed, 11 Jan 2012 09:53:43 +0100
+
 cassandra (1.0.6) unstable; urgency=low
 
   * New release

src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java

@@ -108,7 +108,7 @@ public class DebuggableThreadPoolExecutor extends ThreadPoolExecutor
     protected void onFinalRejection(Runnable task) {}
 
     @Override
-    public void afterExecute(Runnable r, Throwable t)
+    protected void afterExecute(Runnable r, Throwable t)
     {
         super.afterExecute(r,t);
         logExceptionsAfterExecute(r, t);

src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java

@@ -50,6 +50,7 @@ public class NamedThreadFactory implements ThreadFactory
         String name = id + ":" + n.getAndIncrement();
         Thread thread = new Thread(runnable, name);
         thread.setPriority(priority);
+        thread.setDaemon(true);
         return thread;
     }
 }

src/java/org/apache/cassandra/db/Memtable.java

@@ -32,6 +32,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
+import org.apache.cassandra.concurrent.NamedThreadFactory;
 import org.apache.cassandra.db.columniterator.IColumnIterator;
 import org.apache.cassandra.db.columniterator.SimpleAbstractColumnIterator;
 import org.apache.cassandra.db.commitlog.ReplayPosition;
@@ -58,7 +59,12 @@ public class Memtable
     // we're careful to only allow one count to run at a time because counting is slow
     // (can be minutes, for a large memtable and a busy server), so we could keep memtables
     // alive after they're flushed and would otherwise be GC'd.
-    private static final ExecutorService meterExecutor = new ThreadPoolExecutor(1, 1, Integer.MAX_VALUE, TimeUnit.MILLISECONDS, new SynchronousQueue<Runnable>())
+    private static final ExecutorService meterExecutor = new DebuggableThreadPoolExecutor(1,
+                                                                                          1,
+                                                                                          Integer.MAX_VALUE,
+                                                                                          TimeUnit.MILLISECONDS,
+                                                                                          new SynchronousQueue<Runnable>(),
+                                                                                          new NamedThreadFactory("MemoryMeter"))
     {
         @Override
         protected void afterExecute(Runnable r, Throwable t)

src/java/org/apache/cassandra/db/compaction/LeveledManifest.java

@@ -27,6 +27,7 @@ import java.io.IOException;
 import java.util.*;
 
 import com.google.common.collect.Iterables;
+import com.google.common.primitives.Ints;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -203,11 +204,14 @@ public class LeveledManifest
         return builder.toString();
     }
 
-    private double maxBytesForLevel (int level)
+    private long maxBytesForLevel(int level)
     {
-        return level == 0
-               ? 4 * maxSSTableSizeInMB * 1024 * 1024
-               : Math.pow(10, level) * maxSSTableSizeInMB * 1024 * 1024;
+        if (level == 0)
+            return 4 * maxSSTableSizeInMB * 1024 * 1024;
+        double bytes = Math.pow(10, level) * maxSSTableSizeInMB * 1024 * 1024;
+        if (bytes > Long.MAX_VALUE)
+            throw new RuntimeException("At most " + Long.MAX_VALUE + " bytes may be in a compaction level; your maxSSTableSize must be absurdly high to compute " + bytes);
+        return (long) bytes;
     }
 
     public synchronized Collection<SSTableReader> getCompactionCandidates()
@@ -413,12 +417,14 @@ public class LeveledManifest
 
     public int getEstimatedTasks()
     {
-        int n = 0;
+        long tasks = 0;
         for (int i = generations.length - 1; i >= 0; i--)
         {
             List<SSTableReader> sstables = generations[i];
-            n += Math.max(0L, SSTableReader.getTotalBytes(sstables) - maxBytesForLevel(i)) / (maxSSTableSizeInMB * 1024 * 1024);
+            long n = Math.max(0L, SSTableReader.getTotalBytes(sstables) - maxBytesForLevel(i)) / (maxSSTableSizeInMB * 1024 * 1024);
+            logger.debug("Estimating " + n + " compaction tasks in level " + i);
+            tasks += n;
         }
-        return n;
+        return Ints.checkedCast(tasks);
     }
 }

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

@@ -533,15 +533,9 @@ public final class MessagingService implements MessagingServiceMBean
     }
 
     /**
-     * There isn't a good way to shut down the MessagingService. One problem (but not the only one)
-     * is that StorageProxy has no way to communicate back to clients, "I'm nominally alive, but I can't
-     * send that request to the nodes with your data."  Neither TimedOut nor Unavailable is appropriate
-     * to return in that situation.
-     *
-     * So instead of shutting down MS and letting StorageProxy/clients cope somehow, we shut down
-     * the Thrift service and then wait for all the outstanding requests to finish or timeout.
+     * Wait for callbacks and don't allow any more to be created (since they could require writing hints)
      */
-    public void waitForCallbacks()
+    public void shutdown()
     {
         logger_.info("Waiting for messaging service to quiesce");
         // We may need to schedule hints on the mutation stage, so it's erroneous to shut down the mutation stage first

src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java

@@ -30,27 +30,27 @@ import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import com.google.common.collect.Iterables;
 import org.apache.log4j.PropertyConfigurator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
+import org.apache.cassandra.concurrent.NamedThreadFactory;
 import org.apache.cassandra.config.CFMetaData;
 import org.apache.cassandra.config.ConfigurationException;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.config.Schema;
-import org.apache.cassandra.gms.Gossiper;
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.Directories;
 import org.apache.cassandra.db.SystemTable;
 import org.apache.cassandra.db.Table;
 import org.apache.cassandra.db.commitlog.CommitLog;
 import org.apache.cassandra.db.migration.Migration;
+import org.apache.cassandra.gms.Gossiper;
 import org.apache.cassandra.utils.CLibrary;
 import org.apache.cassandra.utils.Mx4jTool;
 
-import com.google.common.collect.Iterables;
-
 /**
  * The <code>CassandraDaemon</code> is an abstraction for a Cassandra daemon
  * service, which defines not only a way to activate and deactivate it, but also
@@ -403,13 +403,16 @@ public abstract class AbstractCassandraDaemon implements CassandraDaemon
     /**
      * A subclass of Java's ThreadPoolExecutor which implements Jetty's ThreadPool
      * interface (for integration with Avro), and performs ClientState cleanup.
+     *
+     * (Note that the tasks being executed perform their own while-command-process
+     * loop until the client disconnects.)
      */
     public static class CleaningThreadPool extends ThreadPoolExecutor 
     {
         private ThreadLocal<ClientState> state;
         public CleaningThreadPool(ThreadLocal<ClientState> state, int minWorkerThread, int maxWorkerThreads)
         {
-            super(minWorkerThread, maxWorkerThreads, 60, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
+            super(minWorkerThread, maxWorkerThreads, 60, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(), new NamedThreadFactory("Thrift"));
             this.state = state;
         }
 
@@ -420,7 +423,5 @@ public abstract class AbstractCassandraDaemon implements CassandraDaemon
             DebuggableThreadPoolExecutor.logExceptionsAfterExecute(r, t);
             state.get().logout();
         }
-
-
     }
 }

src/java/org/apache/cassandra/service/StorageService.java

@@ -327,7 +327,6 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
         daemon.startRPCServer();
     }
 
-    // should only be called via JMX
     public void stopRPCServer()
     {
         if (daemon == null)
@@ -351,7 +350,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
         Gossiper.instance.unregister(migrationManager);
         Gossiper.instance.unregister(this);
         Gossiper.instance.stop();
-        MessagingService.instance().waitForCallbacks();
+        MessagingService.instance().shutdown();
         // give it a second so that task accepted before the MessagingService shutdown gets submitted to the stage (to avoid RejectedExecutionException)
         try { Thread.sleep(1000L); } catch (InterruptedException e) {}
         StageManager.shutdownNow();
@@ -454,7 +453,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
 
                 // In-progress writes originating here could generate hints to be written, so shut down MessagingService
                 // before mutation stage, so we can get all the hints saved before shutting down
-                MessagingService.instance().waitForCallbacks();
+                MessagingService.instance().shutdown();
                 mutationStage.shutdown();
                 mutationStage.awaitTermination(3600, TimeUnit.SECONDS);
                 StorageProxy.instance.verifyNoHintsInProgress();
@@ -2118,7 +2117,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
             public void run()
             {
                 Gossiper.instance.stop();
-                MessagingService.instance().waitForCallbacks();
+                MessagingService.instance().shutdown();
                 StageManager.shutdownNow();
                 setMode(Mode.DECOMMISSIONED, true);
                 // let op be responsible for killing the process
@@ -2520,7 +2519,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
         Gossiper.instance.stop();
 
         setMode(Mode.DRAINING, "shutting down MessageService", false);
-        MessagingService.instance().waitForCallbacks();
+        MessagingService.instance().shutdown();
         setMode(Mode.DRAINING, "waiting for streaming", false);
         MessagingService.instance().waitForStreaming();
 

src/java/org/apache/cassandra/thrift/CustomTThreadPoolServer.java

@@ -119,27 +119,24 @@ public class CustomTThreadPoolServer extends TServer
         }
 
         executorService_.shutdown();
-
-        // Loop until awaitTermination finally does return without a interrupted
-        // exception. If we don't do this, then we'll shut down prematurely. We want
-        // to let the executorService clear it's task queue, closing client sockets
-        // appropriately.
-        long timeoutMS = args.stopTimeoutUnit.toMillis(args.stopTimeoutVal);
-        long now = System.currentTimeMillis();
-        while (timeoutMS >= 0)
-        {
-            try
-            {
-                executorService_.awaitTermination(timeoutMS, TimeUnit.MILLISECONDS);
-                break;
-            }
-            catch (InterruptedException ix)
-            {
-                long newnow = System.currentTimeMillis();
-                timeoutMS -= (newnow - now);
-                now = newnow;
-            }
-        }
+        // Thrift's default shutdown waits for the WorkerProcess threads to complete.  We do not,
+        // because doing that allows a client to hold our shutdown "hostage" by simply not sending
+        // another message after stop is called (since process will block indefinitely trying to read
+        // the next meessage header).
+        //
+        // The "right" fix would be to update thrift to set a socket timeout on client connections
+        // (and tolerate unintentional timeouts until stopped_ is set).  But this requires deep
+        // changes to the code generator, so simply setting these threads to daemon (in our custom
+        // CleaningThreadPool) and ignoring them after shutdown is good enough.
+        //
+        // Remember, our goal on shutdown is not necessarily that each client request we receive
+        // gets answered first [to do that, you should redirect clients to a different coordinator
+        // first], but rather (1) to make sure that for each update we ack as successful, we generate
+        // hints for any non-responsive replicas, and (2) to make sure that we quickly stop
+        // accepting client connections so shutdown can continue.  Not waiting for the WorkerProcess
+        // threads here accomplishes (2); MessagingService's shutdown method takes care of (1).
+        //
+        // See CASSANDRA-3335 and CASSANDRA-3727.
     }
 
     public void stop()
@@ -184,7 +181,9 @@ public class CustomTThreadPoolServer extends TServer
                 inputProtocol = inputProtocolFactory_.getProtocol(inputTransport);
                 outputProtocol = outputProtocolFactory_.getProtocol(outputTransport);
                 // we check stopped_ first to make sure we're not supposed to be shutting
-                // down. this is necessary for graceful shutdown.
+                // down. this is necessary for graceful shutdown.  (but not sufficient,
+                // since process() can take arbitrarily long waiting for client input.
+                // See comments at the end of serve().)
                 while (!stopped_ && processor.process(inputProtocol, outputProtocol))
                 {
                     inputProtocol = inputProtocolFactory_.getProtocol(inputTransport);

src/java/org/apache/cassandra/utils/ExpiringMap.java

@@ -30,6 +30,7 @@ import org.cliffc.high_scale_lib.NonBlockingHashMap;
 public class ExpiringMap<K, V>
 {
     private static final Logger logger = LoggerFactory.getLogger(ExpiringMap.class);
+    private volatile boolean shutdown;
 
     private static class CacheableObject<T>
     {
@@ -104,6 +105,7 @@ public class ExpiringMap<K, V>
 
     public void shutdown()
     {
+        shutdown = true;
         while (!cache.isEmpty())
         {
             logger.trace("Waiting for {} entries before shutting down ExpiringMap", cache.size());
@@ -131,6 +133,21 @@ public class ExpiringMap<K, V>
 
     public V put(K key, V value, long timeout)
     {
+        if (shutdown)
+        {
+            // StorageProxy isn't equipped to deal with "I'm nominally alive, but I can't send any messages out."
+            // So we'll just sit on this thread until the rest of the server shutdown completes.
+            //
+            // See comments in CustomTThreadPoolServer.serve, CASSANDRA-3335, and CASSANDRA-3727.
+            try
+            {
+                Thread.sleep(Long.MAX_VALUE);
+            }
+            catch (InterruptedException e)
+            {
+                throw new AssertionError(e);
+            }
+        }
         CacheableObject<V> previous = cache.put(key, new CacheableObject<V>(value, timeout));
         return (previous == null) ? null : previous.getValue();
     }

test/unit/org/apache/cassandra/service/RemoveTest.java

@@ -85,7 +85,7 @@ public class RemoveTest extends CleanupHelper
     {
         SinkManager.clear();
         MessagingService.instance().clearCallbacksUnsafe();
-        MessagingService.instance().waitForCallbacks();
+        MessagingService.instance().shutdown();
         ss.setPartitionerUnsafe(oldPartitioner);
     }