Repair implements ExecutorTask { private void repairKeyspace(String keyspace, List<String> columnFamilies) throws Exception { LOGGER.info("Starting repair : keySpace = {}, columnFamilies = {}", keyspace, columnFamilies); Map<String, String> options = new HashMap<>(); options.put(RepairOption.PRIMARY_RANGE_KEY, "true"); options.put(RepairOption.COLUMNFAMILIES_KEY, String.join(",", columnFamilies)); options.put(RepairOption.PARALLELISM_KEY, RepairParallelism.SEQUENTIAL.getName()); options.put(RepairOption.INCREMENTAL_KEY, "true"); String result = daemon.repair(keyspace, options); LOGGER.info("Repair output = {}", result); LOGGER.info("Completed repair : keySpace = {}, columnFamilies = {}", keyspace, columnFamilies); sendStatus(driver, Protos.TaskState.TASK_RUNNING, String.format( "Completed repair : keySpace = %s, columnFamilies = %s", keyspace, columnFamilies)); } Repair(final ExecutorDriver driver,
                  final CassandraDaemonProcess daemon,
                  final RepairTask task); @Override void run(); @Override void stop(Future<?> future);  }
@Test public void testRepairKeyspace() throws Exception { repair.run(); ArgumentCaptor<Map> optionsCaptor = ArgumentCaptor.forClass(Map.class); verify(cassandraDaemonProcess).repair(eq("my_keyspace"), optionsCaptor.capture()); Map<String, String> repairOptions = optionsCaptor.getValue(); assertEquals(repairOptions.get(RepairOption.PRIMARY_RANGE_KEY), "true"); assertEquals(repairOptions.get(RepairOption.INCREMENTAL_KEY), "true"); assertEquals(repairOptions.get(RepairOption.COLUMNFAMILIES_KEY), "table1,table2"); }