Fix code quality based on SonarQube report and Mick comments

resource/cassandra-reaper-cassandra2.yaml

@@ -0,0 +1,72 @@
+# Cassandra Reaper Configuration Example.
+# See a bit more complete example in:
+# src/test/resources/cassandra-reaper.yaml
+segmentCount: 200
+repairParallelism: DATACENTER_AWARE
+repairIntensity: 0.9
+scheduleDaysBetween: 7
+repairRunThreadCount: 15
+hangingRepairTimeoutMins: 30
+storageType: cassandra
+enableCrossOrigin: true
+incrementalRepair: false
+allowUnreachableNodes: false
+enableDynamicSeedList: true
+
+jmxPorts:
+  127.0.0.1: 7100
+  127.0.0.2: 7200
+  127.0.0.3: 7300
+  127.0.0.4: 7400
+  127.0.0.5: 7500
+  127.0.0.6: 7600
+  127.0.0.7: 7700
+  127.0.0.8: 7800
+
+logging:
+  level: INFO
+  loggers:
+    com.datastax.driver.core.QueryLogger.NORMAL:
+      level: DEBUG
+      additive: false
+      appenders:
+        - type: file
+          currentLogFilename: query-logger.log
+          archivedLogFilenamePattern: query-logger-%d.log.gz
+          archivedFileCount: 2
+    io.dropwizard: WARN
+    org.eclipse.jetty: WARN
+  appenders:
+    - type: console
+      logFormat: "%-6level [%d] [%t] %logger{5} - %msg %n"
+
+server:
+  type: default
+  applicationConnectors:
+    - type: http
+      port: 8082
+      bindHost: 0.0.0.0
+  adminConnectors:
+    - type: http
+      port: 8083
+      bindHost: 0.0.0.0
+  requestLog:
+    appenders: []
+
+cassandra:
+  clusterName: "test"
+  contactPoints: ["127.0.0.1"]
+  keyspace: reaper_db
+  queryOptions:
+    consistencyLevel: LOCAL_QUORUM
+    serialConsistencyLevel: SERIAL
+
+autoScheduling:
+  enabled: false
+  initialDelayPeriod: PT15S
+  periodBetweenPolls: PT10M
+  timeBeforeFirstSchedule: PT5M
+  scheduleSpreadPeriod: PT6H
+  excludedKeyspaces:
+    - keyspace1
+    - keyspace2

src/main/java/com/spotify/reaper/ReaperApplication.java

@@ -69,9 +69,10 @@
 public class ReaperApplication extends Application<ReaperApplicationConfiguration> {
 
   static final Logger LOG = LoggerFactory.getLogger(ReaperApplication.class);
-  public final static UUID reaperInstanceId = UUID.randomUUID();
+  public static final UUID REAPER_INSTANCE_ID = UUID.randomUUID();
   private static String reaperInstanceAddress;
   private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
+  private static final String DEFAULT_INSTANCE_ADDRESS = "127.0.0.1";
 
   private AppContext context;
 
@@ -94,12 +95,12 @@ public static void main(String[] args) throws Exception {
     new ReaperApplication().run(args);
   }
 
-  private void setInstanceAddress() {
+  private static void setInstanceAddress() {
     try{
-      this.reaperInstanceAddress = InetAddress.getLocalHost().getHostAddress();
+      ReaperApplication.reaperInstanceAddress = InetAddress.getLocalHost().getHostAddress();
     } catch(Exception e) {
       LOG.warn("Cannot get instance address", e);
-      this.reaperInstanceAddress = "127.0.0.1";
+      ReaperApplication.reaperInstanceAddress = DEFAULT_INSTANCE_ADDRESS;
     }
   }
 
@@ -219,14 +220,12 @@ public void run(ReaperApplicationConfiguration config,
       // us to poll the database for running repairs regularly
       // only with Cassandra storage
       scheduler.scheduleWithFixedDelay(
-        new Runnable() {
-          public void run() { 
+        () -> {
             try {
               context.repairManager.resumeRunningRepairRuns(context);
             } catch (ReaperException e) {
               LOG.error("Couldn't resume running repair runs", e);
             } 
-          }
         }, 0, 10, TimeUnit.SECONDS);
     }
     else {

src/main/java/com/spotify/reaper/core/HostMetrics.java

@@ -1,8 +1,6 @@
 package com.spotify.reaper.core;
 
-import javax.annotation.Generated;
-
-public class HostMetrics {
+public final class HostMetrics {
   private final String hostAddress;
   private final int pendingCompactions;
   private final boolean hasRepairRunning;

src/main/java/com/spotify/reaper/storage/CassandraStorage.java

@@ -781,7 +781,7 @@ private RepairRun buildRepairRunFromRow(Row repairRunResult, long id){
 
   @Override
   public boolean takeLeadOnSegment(long segmentId) {
-    Row lwtResult = session.execute(getLeadOnSegmentPrepStmt.bind(segmentId, ReaperApplication.reaperInstanceId, ReaperApplication.getInstanceAddress())).one();
+    Row lwtResult = session.execute(getLeadOnSegmentPrepStmt.bind(segmentId, ReaperApplication.REAPER_INSTANCE_ID, ReaperApplication.getInstanceAddress())).one();
     if (lwtResult.getBool("[applied]")) {
       LOG.debug("Took lead on segment {}", segmentId);
       return true;
@@ -794,7 +794,7 @@ public boolean takeLeadOnSegment(long segmentId) {
 
   @Override
   public boolean renewLeadOnSegment(long segmentId) {
-    Row lwtResult = session.execute(renewLeadOnSegmentPrepStmt.bind(ReaperApplication.reaperInstanceId, ReaperApplication.getInstanceAddress(), segmentId, ReaperApplication.reaperInstanceId)).one();
+    Row lwtResult = session.execute(renewLeadOnSegmentPrepStmt.bind(ReaperApplication.REAPER_INSTANCE_ID, ReaperApplication.getInstanceAddress(), segmentId, ReaperApplication.REAPER_INSTANCE_ID)).one();
     if (lwtResult.getBool("[applied]")) {
       LOG.debug("Renewed lead on segment {}", segmentId);
       return true;
@@ -804,8 +804,12 @@ public boolean renewLeadOnSegment(long segmentId) {
 
   @Override
   public void releaseLeadOnSegment(long segmentId) {
-    Row lwtResult = session.execute(releaseLeadOnSegmentPrepStmt.bind(segmentId, ReaperApplication.reaperInstanceId)).one();
-    LOG.debug("Released lead on segment {}", segmentId);
+    Row lwtResult = session.execute(releaseLeadOnSegmentPrepStmt.bind(segmentId, ReaperApplication.REAPER_INSTANCE_ID)).one();
+    if (lwtResult.getBool("[applied]")) {
+      LOG.debug("Released lead on segment {}", segmentId);
+    } else {
+      LOG.error("Could not release lead on segment {}", segmentId);
+    }
   }
 
   @Override
@@ -845,7 +849,7 @@ public void saveHeartbeat() {
     DateTime now = DateTime.now();
     // Send heartbeats every minute
     if(now.minusSeconds(60).getMillis() >= lastHeartBeat.getMillis()) {
-      session.executeAsync(saveHeartbeatPrepStmt.bind(ReaperApplication.reaperInstanceId, ReaperApplication.getInstanceAddress()));
+      session.executeAsync(saveHeartbeatPrepStmt.bind(ReaperApplication.REAPER_INSTANCE_ID, ReaperApplication.getInstanceAddress()));
       lastHeartBeat = now;
     }
   }

src/main/java/com/spotify/reaper/storage/IStorage.java

@@ -26,7 +26,6 @@
 import com.spotify.reaper.service.RepairParameters;
 import com.spotify.reaper.service.RingRange;
 
-import java.net.UnknownHostException;
 import java.util.Collection;
 import java.util.Set;
 

src/main/java/com/spotify/reaper/storage/MemoryStorage.java

@@ -509,10 +509,12 @@ public boolean renewLeadOnSegment(long segmentId) {
 
   @Override
   public void releaseLeadOnSegment(long segmentId) {
+    // Fault tolerance is not supported with this storage backend
   }
 
   @Override
   public void storeHostMetrics(HostMetrics hostMetrics) {
+    // Fault tolerance is not supported with this storage backend
   }
 
   @Override
@@ -532,5 +534,6 @@ public int countRunningReapers() {
 
   @Override
   public void saveHeartbeat() {
+    // Fault tolerance is not supported with this storage backend
   }
 }

src/main/java/com/spotify/reaper/storage/PostgresStorage.java

@@ -527,5 +527,6 @@ public int countRunningReapers() {
 
   @Override
   public void saveHeartbeat() {
+    // Fault tolerance is not supported with this storage backend
   }
 }

src/main/resources/db/cassandra/002_fault_tolerant_reaper.cql → src/main/resources/db/cassandra/003_fault_tolerant_reaper.cql

@@ -16,3 +16,13 @@ CREATE TABLE IF NOT EXISTS host_metrics(
   active_anticompactions int,
   PRIMARY KEY(host_address)
 ) WITH default_time_to_live=180;
+
+--
+-- Add table to track concurrently running Reapers
+
+CREATE TABLE IF NOT EXISTS running_reapers (
+  reaper_instance_id uuid PRIMARY KEY,
+  reaper_instance_host text,
+	last_heartbeat timestamp
+) WITH default_time_to_live = 180
+  AND gc_grace_seconds = 10800;