In CassandraStorage implement segments as clustering keys within the …

…repair_run table. (#102)

Change required in IStorage so to identify a segment both by runId and segmentId.

ref:
 - #94
 - #102

pom.xml

@@ -11,7 +11,7 @@
 
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <dropwizard.version>1.1.0</dropwizard.version>
+        <dropwizard.version>1.0.7</dropwizard.version>
         <dropwizard.cassandra.version>4.1.0</dropwizard.cassandra.version>
         <cassandra.version>2.2.7</cassandra.version>
         <cucumber.version>1.2.5</cucumber.version>

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

@@ -254,7 +254,7 @@ public void setCassandraFactory(CassandraFactory cassandra) {
   }
 
   public Boolean getAllowUnreachableNodes() {
-    return allowUnreachableNodes;
+    return allowUnreachableNodes != null ? allowUnreachableNodes : false;
   }
 
   public void setAllowUnreachableNodes(Boolean allow) {

src/main/java/com/spotify/reaper/service/RepairRunner.java

@@ -229,7 +229,7 @@ private void startNextSegment() throws ReaperException {
 
         // Just checking that no currently running segment runner is stuck.
         RepairSegment supposedlyRunningSegment =
-            context.storage.getRepairSegment(currentlyRunningSegments.get(rangeIndex)).get();
+            context.storage.getRepairSegment(repairRunId, currentlyRunningSegments.get(rangeIndex)).get();
         DateTime startTime = supposedlyRunningSegment.getStartTime();
         if (startTime != null && startTime.isBefore(DateTime.now().minusDays(1))) {
           LOG.warn("Looks like segment #{} has been running more than a day. Start time: {}",
@@ -343,7 +343,8 @@ private boolean repairSegment(final int rangeIndex, final UUID segmentId, RingRa
 	    }
     } 
     else {
-    	potentialCoordinators = Arrays.asList(context.storage.getRepairSegment(segmentId).get().getCoordinatorHost());
+    	potentialCoordinators 
+                = Arrays.asList(context.storage.getRepairSegment(repairRunId, segmentId).get().getCoordinatorHost());
     }
 
     SegmentRunner segmentRunner = new SegmentRunner(context, segmentId, potentialCoordinators,
@@ -369,7 +370,7 @@ public void onFailure(Throwable t) {
   }
 
   private void handleResult(UUID segmentId) {
-    RepairSegment segment = context.storage.getRepairSegment(segmentId).get();
+    RepairSegment segment = context.storage.getRepairSegment(repairRunId, segmentId).get();
     RepairSegment.State segmentState = segment.getState();
     LOG.debug("In repair run #{}, triggerRepair on segment {} ended with state {}",
         repairRunId, segmentId, segmentState);

src/main/java/com/spotify/reaper/service/SegmentRunner.java

@@ -79,6 +79,8 @@ public final class SegmentRunner implements RepairStatusHandler, Runnable {
   public SegmentRunner(AppContext context, UUID segmentId, Collection<String> potentialCoordinators,
       long timeoutMillis, double intensity, RepairParallelism validationParallelism,
       String clusterName, RepairUnit repairUnit, RepairRunner repairRunner) {
+
+    assert !segmentRunners.containsKey(segmentId) : "SegmentRunner already exists for segment with ID: " + segmentId;
     this.context = context;
     this.segmentId = segmentId;
     this.potentialCoordinators = potentialCoordinators;
@@ -93,7 +95,7 @@ public SegmentRunner(AppContext context, UUID segmentId, Collection<String> pote
 
   @Override
   public void run() {
-    final RepairSegment segment = context.storage.getRepairSegment(segmentId).get();
+    final RepairSegment segment = context.storage.getRepairSegment(repairRunner.getRepairRunId(), segmentId).get();
     Thread.currentThread().setName(clusterName + ":" + segment.getRunId() + ":" + segmentId);
 
     runRepair();
@@ -129,7 +131,7 @@ public static void abort(AppContext context, RepairSegment segment, JmxProxy jmx
    */
   public void postponeCurrentSegment() {
     synchronized (condition) {
-      RepairSegment segment = context.storage.getRepairSegment(segmentId).get();
+      RepairSegment segment = context.storage.getRepairSegment(repairRunner.getRepairRunId(), segmentId).get();
       postpone(context, segment, context.storage.getRepairUnit(segment.getRepairUnitId()));
     }
   }
@@ -149,7 +151,7 @@ private long getOpenFilesAmount() {
 
   private void runRepair() {
     LOG.debug("Run repair for segment #{}", segmentId);
-    final RepairSegment segment = context.storage.getRepairSegment(segmentId).get();
+    final RepairSegment segment = context.storage.getRepairSegment(repairRunner.getRepairRunId(), segmentId).get();
     try (JmxProxy coordinator = context.jmxConnectionFactory
         .connectAny(Optional.<RepairStatusHandler>fromNullable(this), potentialCoordinators)) {
 
@@ -220,7 +222,7 @@ protected Set<String> initialize() {
         } catch (InterruptedException e) {
           LOG.warn("Repair command {} on segment {} interrupted", commandId, segmentId, e);
         } finally {
-          RepairSegment resultingSegment = context.storage.getRepairSegment(segmentId).get();
+          RepairSegment resultingSegment = context.storage.getRepairSegment(repairRunner.getRepairRunId(), segmentId).get();
           LOG.info("Repair command {} on segment {} returned with state {}", commandId, segmentId,
               resultingSegment.getState());
           if (resultingSegment.getState() == RepairSegment.State.RUNNING) {
@@ -382,7 +384,7 @@ private void abort(RepairSegment segment, JmxProxy jmxConnection) {
    */
   @Override
   public void handle(int repairNumber, Optional<ActiveRepairService.Status> status, Optional<ProgressEventType> progress, String message) {
-    final RepairSegment segment = context.storage.getRepairSegment(segmentId).get();
+    final RepairSegment segment = context.storage.getRepairSegment(repairRunner.getRepairRunId(), segmentId).get();
     Thread.currentThread().setName(clusterName + ":" + segment.getRunId() + ":" + segmentId);
     LOG.debug(
         "handle called for repairCommandId {}, outcome {} / {} and message: {}",
@@ -396,7 +398,7 @@ public void handle(int repairNumber, Optional<ActiveRepairService.Status> status
     boolean failOutsideSynchronizedBlock = false;
     // DO NOT ADD EXTERNAL CALLS INSIDE THIS SYNCHRONIZED BLOCK (JMX PROXY ETC)
     synchronized (condition) {
-      RepairSegment currentSegment = context.storage.getRepairSegment(segmentId).get();
+      RepairSegment currentSegment = context.storage.getRepairSegment(repairRunner.getRepairRunId(), segmentId).get();
       // See status explanations at: https://wiki.apache.org/cassandra/RepairAsyncAPI
       // Old repair API
       if(status.isPresent()) {
@@ -524,7 +526,7 @@ public static String parseRepairId(String message) {
    * @return the delay in milliseconds.
    */
   long intensityBasedDelayMillis(double intensity) {
-    RepairSegment repairSegment = context.storage.getRepairSegment(segmentId).get();
+    RepairSegment repairSegment = context.storage.getRepairSegment(repairRunner.getRepairRunId(), segmentId).get();
     if (repairSegment.getEndTime() == null && repairSegment.getStartTime() == null) {
       return 0;
     } else if (repairSegment.getEndTime() != null && repairSegment.getStartTime() != null) {