Merge branch 'cassandra-1.2' into cassandra-2.0

Conflicts:
	NEWS.txt
	src/java/org/apache/cassandra/db/BatchlogManager.java

CHANGES.txt

@@ -3,6 +3,7 @@
 Merged from 1.2:
  * fsync compression metadata (CASSANDRA-6531)
  * Validate CF existence on execution for prepared statement (CASSANDRA-6535)
+ * Add ability to throttle batchlog replay (CASSANDRA-6550)
 
 
 2.0.4

NEWS.txt

@@ -13,6 +13,16 @@ restore snapshots created with the previous major version using the
 'sstableloader' tool. You can upgrade the file format of your snapshots
 using the provided 'sstableupgrade' tool.
 
+
+2.0.5
+=====
+
+New features
+--------
+    - Batchlog replay can be, and is throttled by default now.
+      See batchlog_replay_throttle_in_kb setting in cassandra.yaml.
+
+
 2.0.3
 =====
 

conf/cassandra.yaml

@@ -46,6 +46,10 @@ hinted_handoff_throttle_in_kb: 1024
 # cross-dc handoff tends to be slower
 max_hints_delivery_threads: 2
 
+# Maximum throttle in KBs per second, total. This will be
+# reduced proportionally to the number of nodes in the cluster.
+batchlog_replay_throttle_in_kb: 1024
+
 # The following setting populates the page cache on memtable flush and compaction
 # WARNING: Enable this setting only when the whole node's data fits in memory.
 # Defaults to: false

src/java/org/apache/cassandra/config/Config.java

@@ -150,6 +150,7 @@ public class Config
     public Integer index_interval = null;
 
     public int hinted_handoff_throttle_in_kb = 1024;
+    public int batchlog_replay_throttle_in_kb = 1024;
     public int max_hints_delivery_threads = 1;
     public boolean compaction_preheat_key_cache = true;
 

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -1170,6 +1170,11 @@ public class DatabaseDescriptor
         return conf.hinted_handoff_throttle_in_kb;
     }
 
+    public static int getBatchlogReplayThrottleInKB()
+    {
+        return conf.batchlog_replay_throttle_in_kb;
+    }
+
     public static int getMaxHintsThread()
     {
         return conf.max_hints_delivery_threads;

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

@@ -35,6 +35,7 @@ import javax.management.ObjectName;
 
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.RateLimiter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -75,7 +76,7 @@ public class BatchlogManager implements BatchlogManagerMBean
     private final AtomicBoolean isReplaying = new AtomicBoolean();
 
     private static final ScheduledExecutorService batchlogTasks = new DebuggableScheduledThreadPoolExecutor("BatchlogTasks");
-    
+
     public void start()
     {
         MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
@@ -161,11 +162,16 @@ public class BatchlogManager implements BatchlogManagerMBean
 
         logger.debug("Started replayAllFailedBatches");
 
+        // rate limit is in bytes per second. Uses Double.MAX_VALUE if disabled (set to 0 in cassandra.yaml).
+        // max rate is scaled by the number of nodes in the cluster (same as for HHOM - see CASSANDRA-5272).
+        int throttleInKB = DatabaseDescriptor.getBatchlogReplayThrottleInKB() / StorageService.instance.getTokenMetadata().getAllEndpoints().size();
+        RateLimiter rateLimiter = RateLimiter.create(throttleInKB == 0 ? Double.MAX_VALUE : throttleInKB * 1024);
+
         try
         {
             for (UntypedResultSet.Row row : process("SELECT id, written_at FROM %s.%s", Keyspace.SYSTEM_KS, SystemKeyspace.BATCHLOG_CF))
                 if (System.currentTimeMillis() > row.getLong("written_at") + TIMEOUT)
-                    replayBatch(row.getUUID("id"));
+                    replayBatch(row.getUUID("id"), rateLimiter);
             cleanup();
         }
         finally
@@ -176,7 +182,7 @@ public class BatchlogManager implements BatchlogManagerMBean
         logger.debug("Finished replayAllFailedBatches");
     }
 
-    private void replayBatch(UUID id)
+    private void replayBatch(UUID id, RateLimiter rateLimiter)
     {
         logger.debug("Replaying batch {}", id);
 
@@ -187,7 +193,7 @@ public class BatchlogManager implements BatchlogManagerMBean
         try
         {
             UntypedResultSet.Row batch = result.one();
-            replaySerializedMutations(batch.getBytes("data"), batch.getLong("written_at"));
+            replaySerializedMutations(batch.getBytes("data"), batch.getLong("written_at"), rateLimiter);
         }
         catch (IOException e)
         {
@@ -199,30 +205,33 @@ public class BatchlogManager implements BatchlogManagerMBean
         totalBatchesReplayed.incrementAndGet();
     }
 
-    private void replaySerializedMutations(ByteBuffer data, long writtenAt) throws IOException
+    private void replaySerializedMutations(ByteBuffer data, long writtenAt, RateLimiter rateLimiter) throws IOException
     {
         DataInputStream in = new DataInputStream(ByteBufferUtil.inputStream(data));
         int size = in.readInt();
         for (int i = 0; i < size; i++)
-            replaySerializedMutation(RowMutation.serializer.deserialize(in, VERSION), writtenAt);
+            replaySerializedMutation(RowMutation.serializer.deserialize(in, VERSION), writtenAt, rateLimiter);
     }
 
     /*
      * We try to deliver the mutations to the replicas ourselves if they are alive and only resort to writing hints
      * when a replica is down or a write request times out.
      */
-    private void replaySerializedMutation(RowMutation mutation, long writtenAt)
+    private void replaySerializedMutation(RowMutation mutation, long writtenAt, RateLimiter rateLimiter)
     {
         int ttl = calculateHintTTL(mutation, writtenAt);
         if (ttl <= 0)
             return; // the mutation isn't safe to replay.
 
-        Set<InetAddress> liveEndpoints = new HashSet<InetAddress>();
+        Set<InetAddress> liveEndpoints = new HashSet<>();
         String ks = mutation.getKeyspaceName();
         Token<?> tk = StorageService.getPartitioner().getToken(mutation.key());
+        int mutationSize = (int) RowMutation.serializer.serializedSize(mutation, VERSION);
+
         for (InetAddress endpoint : Iterables.concat(StorageService.instance.getNaturalEndpoints(ks, tk),
                                                      StorageService.instance.getTokenMetadata().pendingEndpointsFor(tk, ks)))
         {
+            rateLimiter.acquire(mutationSize);
             if (endpoint.equals(FBUtilities.getBroadcastAddress()))
                 mutation.apply();
             else if (FailureDetector.instance.isAlive(endpoint))
@@ -238,7 +247,7 @@ public class BatchlogManager implements BatchlogManagerMBean
     private void attemptDirectDelivery(RowMutation mutation, long writtenAt, Set<InetAddress> endpoints)
     {
         List<WriteResponseHandler> handlers = Lists.newArrayList();
-        final CopyOnWriteArraySet<InetAddress> undelivered = new CopyOnWriteArraySet<InetAddress>(endpoints);
+        final CopyOnWriteArraySet<InetAddress> undelivered = new CopyOnWriteArraySet<>(endpoints);
         for (final InetAddress ep : endpoints)
         {
             Runnable callback = new Runnable()
@@ -292,7 +301,7 @@ public class BatchlogManager implements BatchlogManagerMBean
     {
         ColumnFamilyStore cfs = Keyspace.open(Keyspace.SYSTEM_KS).getColumnFamilyStore(SystemKeyspace.BATCHLOG_CF);
         cfs.forceBlockingFlush();
-        Collection<Descriptor> descriptors = new ArrayList<Descriptor>();
+        Collection<Descriptor> descriptors = new ArrayList<>();
         for (SSTableReader sstr : cfs.getSSTables())
             descriptors.add(sstr.descriptor);
         if (!descriptors.isEmpty()) // don't pollute the logs if there is nothing to compact.