From 1492be07cdf12afbba662645365cb991f223443d Mon Sep 17 00:00:00 2001
From: Branimir Lambov <branimir.lambov@datastax.com>
Date: Wed, 28 Oct 2015 17:11:35 +0200
Subject: [PATCH] Guard batchlog replay against integer division by zero

patch by Branimir Lambov; reviewed by Stefania Alborghetti for
CASSANDRA-9223
---
 CHANGES.txt                                   |  1 +
 .../cassandra/batchlog/BatchlogManager.java   |  8 +++-
 .../batchlog/BatchlogManagerTest.java         | 40 +++++++++++++++++++
 3 files changed, 48 insertions(+), 1 deletion(-)

diff --git a/CHANGES.txt b/CHANGES.txt
index dfb0f1729c..61cc92e4a4 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 3.0
+ * Guard batchlog replay against integer division by zero (CASSANDRA-9223)
  * Fix bug when adding a column to thrift with the same name than a primary key (CASSANDRA-10608)
  * Add client address argument to IAuthenticator::newSaslNegotiator (CASSANDRA-8068)
  * Fix implementation of LegacyLayout.LegacyBoundComparator (CASSANDRA-10602)
diff --git a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
index 8bc4c269d7..1c98c2a303 100644
--- a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
+++ b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
@@ -178,7 +178,13 @@ public class BatchlogManager implements BatchlogManagerMBean
 
         // 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();
+        int endpointsCount = StorageService.instance.getTokenMetadata().getAllEndpoints().size();
+        if (endpointsCount <= 0)
+        {
+            logger.trace("Replay cancelled as there are no peers in the ring.");
+            return;
+        }
+        int throttleInKB = DatabaseDescriptor.getBatchlogReplayThrottleInKB() / endpointsCount;
         RateLimiter rateLimiter = RateLimiter.create(throttleInKB == 0 ? Double.MAX_VALUE : throttleInKB * 1024);
 
         UUID limitUuid = UUIDGen.maxTimeUUID(System.currentTimeMillis() - getBatchlogTimeout());
diff --git a/test/unit/org/apache/cassandra/batchlog/BatchlogManagerTest.java b/test/unit/org/apache/cassandra/batchlog/BatchlogManagerTest.java
index dfb17c3de8..dd5444fac9 100644
--- a/test/unit/org/apache/cassandra/batchlog/BatchlogManagerTest.java
+++ b/test/unit/org/apache/cassandra/batchlog/BatchlogManagerTest.java
@@ -24,6 +24,7 @@ import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
 import com.google.common.collect.Lists;
+
 import org.junit.*;
 
 import org.apache.cassandra.SchemaLoader;
@@ -457,4 +458,43 @@ public class BatchlogManagerTest
         assertNotNull(result);
         assertEquals(0L, result.one().getLong("count"));
     }
+
+    // CASSANRDA-9223
+    @Test
+    public void testReplayWithNoPeers() throws Exception
+    {
+        StorageService.instance.getTokenMetadata().removeEndpoint(InetAddress.getByName("127.0.0.1"));
+
+        long initialAllBatches = BatchlogManager.instance.countAllBatches();
+        long initialReplayedBatches = BatchlogManager.instance.getTotalBatchesReplayed();
+
+        CFMetaData cfm = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD1).metadata;
+
+        long timestamp = (System.currentTimeMillis() - DatabaseDescriptor.getWriteRpcTimeout() * 2) * 1000;
+        UUID uuid = UUIDGen.getTimeUUID();
+
+        // Add a batch with 10 mutations
+        List<Mutation> mutations = new ArrayList<>(10);
+        for (int j = 0; j < 10; j++)
+        {
+            mutations.add(new RowUpdateBuilder(cfm, FBUtilities.timestampMicros(), ByteBufferUtil.bytes(j))
+                          .clustering("name" + j)
+                          .add("val", "val" + j)
+                          .build());
+        }
+        BatchlogManager.store(Batch.createLocal(uuid, timestamp, mutations));
+        assertEquals(1, BatchlogManager.instance.countAllBatches() - initialAllBatches);
+
+        // Flush the batchlog to disk (see CASSANDRA-6822).
+        Keyspace.open(SystemKeyspace.NAME).getColumnFamilyStore(SystemKeyspace.BATCHES).forceBlockingFlush();
+
+        assertEquals(1, BatchlogManager.instance.countAllBatches() - initialAllBatches);
+        assertEquals(0, BatchlogManager.instance.getTotalBatchesReplayed() - initialReplayedBatches);
+
+        // Force batchlog replay and wait for it to complete.
+        BatchlogManager.instance.startBatchlogReplay().get();
+
+        // Replay should be cancelled as there are no peers in the ring.
+        assertEquals(1, BatchlogManager.instance.countAllBatches() - initialAllBatches);
+    }
 }