CEP-15: C* - Early repair failures hang nodetool

Patch by Blake Eggleston; Reviewed by Ariel Weisberg for CASSANDRA-19834

src/java/org/apache/cassandra/service/StorageService.java

@@ -169,6 +169,7 @@ import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.schema.TableMetadataRef;
 import org.apache.cassandra.schema.ViewMetadata;
 import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState;
+import org.apache.cassandra.service.consensus.migration.ConsensusMigrationTarget;
 import org.apache.cassandra.service.disk.usage.DiskUsageBroadcaster;
 import org.apache.cassandra.service.paxos.Paxos;
 import org.apache.cassandra.service.paxos.PaxosCommit;
@@ -1686,12 +1687,13 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
     }
 
     @Override
-    public List<Integer> finishConsensusMigration(@Nonnull String keyspace,
+    public Integer finishConsensusMigration(@Nonnull String keyspace,
-                                                  @Nullable List<String> maybeTableNames,
+                                            @Nullable List<String> maybeTableNames,
-                                                  @Nullable String maybeRangesStr)
+                                            @Nullable String maybeRangesStr,
+                                            @Nonnull ConsensusMigrationTarget target)
     {
         checkArgument(!keyspace.equals(SchemaConstants.METADATA_KEYSPACE_NAME));
-        return finishMigrationToConsensusProtocol(keyspace, Optional.ofNullable(maybeTableNames), Optional.ofNullable(maybeRangesStr));
+        return finishMigrationToConsensusProtocol(keyspace, Optional.ofNullable(maybeTableNames), Optional.ofNullable(maybeRangesStr), target);
     }
 
     @Override

src/java/org/apache/cassandra/service/StorageServiceMBean.java

@@ -36,6 +36,7 @@ import javax.management.openmbean.TabularData;
 
 import org.apache.cassandra.db.ColumnFamilyStoreMBean;
 import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.service.consensus.migration.ConsensusMigrationTarget;
 import org.apache.cassandra.utils.BreaksJMX;
 
 public interface StorageServiceMBean extends NotificationEmitter
@@ -1147,9 +1148,10 @@ public interface StorageServiceMBean extends NotificationEmitter
                                   @Nullable List<String> maybeTableNames,
                                   @Nullable String maybeRangesStr);
 
-    List<Integer> finishConsensusMigration(@Nonnull String keyspace,
+    Integer finishConsensusMigration(@Nonnull String keyspace,
-                                           @Nullable List<String> maybeTableNames,
+                                     @Nullable List<String> maybeTableNames,
-                                           @Nullable String maybeRangesStr);
+                                     @Nullable String maybeRangesStr,
+                                     @Nonnull ConsensusMigrationTarget target);
 
     String listConsensusMigrations(@Nullable Set<String> keyspaceNames, @Nullable Set<String> tableNames, @Nonnull String format);
 

src/java/org/apache/cassandra/service/consensus/migration/ConsensusTableMigration.java

@@ -181,11 +181,13 @@ public abstract class ConsensusTableMigration
         ClusterMetadataService.instance().commit(new BeginConsensusMigrationForTableAndRange(targetProtocol, ranges, tableIds));
     }
 
-    public static List<Integer> finishMigrationToConsensusProtocol(@Nonnull String keyspace,
+    public static Integer finishMigrationToConsensusProtocol(@Nonnull String keyspace,
                                                                    @Nonnull Optional<List<String>> maybeTables,
-                                                                   @Nonnull Optional<String> maybeRangesStr)
+                                                                   @Nonnull Optional<String> maybeRangesStr,
+                                                                   ConsensusMigrationTarget target)
     {
         checkArgument(!maybeTables.isPresent() || !maybeTables.get().isEmpty(), "Must provide at least 1 table if Optional is not empty");
+        checkNotNull(target);
 
         Optional<List<Range<Token>>> localKeyspaceRanges = Optional.of(ImmutableList.copyOf(StorageService.instance.getLocalReplicas(keyspace).onlyFull().ranges()));
         List<Range<Token>> ranges = maybeRangesToRanges(maybeRangesStr, localKeyspaceRanges);
@@ -215,17 +217,17 @@ public abstract class ConsensusTableMigration
             tableMigrationStates.add(tms);
         });
 
-        List<TableMigrationState> migratingToAccord = tableMigrationStates.stream().filter(tms -> tms.targetProtocol == ConsensusMigrationTarget.accord).collect(toImmutableList());
+        switch (target)
-        List<TableMigrationState> migratingToPaxos = tableMigrationStates.stream().filter(tms -> tms.targetProtocol == ConsensusMigrationTarget.paxos).collect(toImmutableList());;
+        {
-
+            case accord:
-        Integer accordRepairCmd = finishMigrationToAccord(keyspace, migratingToAccord, ranges);
+                List<TableMigrationState> migratingToAccord = tableMigrationStates.stream().filter(tms -> tms.targetProtocol == ConsensusMigrationTarget.accord).collect(toImmutableList());
-        Integer paxosRepairCmd = finishMigrationToPaxos(keyspace, migratingToPaxos, ranges);
+                return finishMigrationToAccord(keyspace, migratingToAccord, ranges);
-        List<Integer> result = new ArrayList<>();
+            case paxos:
-        if (accordRepairCmd != null)
+                List<TableMigrationState> migratingToPaxos = tableMigrationStates.stream().filter(tms -> tms.targetProtocol == ConsensusMigrationTarget.paxos).collect(toImmutableList());;
-            result.add(accordRepairCmd);
+                return finishMigrationToPaxos(keyspace, migratingToPaxos, ranges);
-        if (paxosRepairCmd != null)
+            default:
-            result.add(paxosRepairCmd);
+                throw new IllegalArgumentException("Unsupported target: " + target);
-        return result;
+        }
     }
 
     private interface MigrationFinisher

src/java/org/apache/cassandra/tools/NodeProbe.java

@@ -30,6 +30,7 @@ import java.rmi.server.RMIClientSocketFactory;
 import java.rmi.server.RMISocketFactory;
 import java.util.AbstractMap;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
@@ -114,7 +115,6 @@ import org.apache.cassandra.service.ActiveRepairServiceMBean;
 import org.apache.cassandra.service.CacheService;
 import org.apache.cassandra.service.CacheServiceMBean;
 import org.apache.cassandra.service.snapshot.SnapshotManagerMBean;
-import org.apache.cassandra.tcm.CMSOperationsMBean;
 import org.apache.cassandra.service.GCInspector;
 import org.apache.cassandra.service.GCInspectorMXBean;
 import org.apache.cassandra.service.StorageProxy;
@@ -126,6 +126,8 @@ import org.apache.cassandra.streaming.StreamManagerMBean;
 import org.apache.cassandra.streaming.StreamState;
 import org.apache.cassandra.streaming.management.StreamStateCompositeData;
 import org.apache.cassandra.tcm.CMSOperations;
+import org.apache.cassandra.tcm.CMSOperationsMBean;
+import org.apache.cassandra.tools.RepairRunner.RepairCmd;
 import org.apache.cassandra.tools.nodetool.GetTimeout;
 import org.apache.cassandra.tools.nodetool.formatter.TableBuilder;
 import org.apache.cassandra.utils.NativeLibrary;
@@ -525,18 +527,28 @@ public class NodeProbe implements AutoCloseable
 
     public void repairAsync(final PrintStream out, final String keyspace, Map<String, String> options) throws IOException
     {
-        blockOnAsyncRepair(out, keyspace, ssProxy.repairAsync(keyspace, options));
+        startAndBlockOnAsyncRepairs(out, Collections.singleton(new RepairCmd(keyspace)
+        {
+            @Override
+            public Integer start()
+            {
+                return ssProxy.repairAsync(keyspace, options);
+            }
+        }));
     }
 
-    public void blockOnAsyncRepair(final PrintStream out, final String keyspace, Integer cmd) throws IOException
+    public void startAndBlockOnAsyncRepairs(final PrintStream out, Collection<RepairCmd> cmds) throws IOException
     {
-        RepairRunner runner = new RepairRunner(out, ssProxy, keyspace, cmd);
+        List<RepairRunner> runners = new ArrayList<>(cmds.size());
+        for (RepairCmd cmd : cmds)
+            runners.add(new RepairRunner(out, jmxc, ssProxy, cmd));
+
         try
         {
-            if (jmxc != null)
+            runners.forEach(RepairRunner::start);
-                jmxc.addConnectionNotificationListener(runner, null, null);
+
-            ssProxy.addNotificationListener(runner, null, null);
+            for (RepairRunner runner : runners)
-            runner.run();
+                runner.run();
         }
         catch (Exception e)
         {
@@ -546,9 +558,7 @@ public class NodeProbe implements AutoCloseable
         {
             try
             {
-                ssProxy.removeNotificationListener(runner);
+                runners.forEach(RepairRunner::close);
-                if (jmxc != null)
-                    jmxc.removeConnectionNotificationListener(runner);
             }
             catch (Throwable e)
             {

src/java/org/apache/cassandra/tools/RepairRunner.java

@@ -22,6 +22,9 @@ import java.io.PrintStream;
 import java.text.SimpleDateFormat;
 import java.util.List;
 
+import javax.management.ListenerNotFoundException;
+import javax.management.remote.JMXConnector;
+
 import org.apache.cassandra.service.ActiveRepairService.ParentRepairStatus;
 import org.apache.cassandra.service.StorageServiceMBean;
 import org.apache.cassandra.utils.concurrent.Condition;
@@ -42,30 +45,78 @@ import static org.apache.cassandra.utils.progress.ProgressEventType.PROGRESS;
 
 public class RepairRunner extends JMXNotificationProgressListener
 {
+    public static abstract class RepairCmd
+    {
+        private final String keyspace;
+
+        public RepairCmd(String keyspace)
+        {
+            this.keyspace = keyspace;
+        }
+
+
+
+        public abstract Integer start();
+    }
     private final SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss,SSS");
 
     private final PrintStream out;
+    private final JMXConnector jmxc;
     private final StorageServiceMBean ssProxy;
-    private final String keyspace;
     private final Condition condition = newOneTimeCondition();
 
+    private final RepairCmd repairCmd;
     private Integer cmd;
     private volatile Exception error;
 
-    public RepairRunner(PrintStream out, StorageServiceMBean ssProxy, String keyspace, Integer cmd)
+    public RepairRunner(PrintStream out, JMXConnector jmxc, StorageServiceMBean ssProxy, RepairCmd repairCmd)
     {
         this.out = out;
+        this.jmxc = jmxc;
         this.ssProxy = ssProxy;
-        this.keyspace = keyspace;
+        this.repairCmd = repairCmd;
-        this.cmd = cmd;
+    }
+
+    public void start()
+    {
+        if (jmxc != null)
+            jmxc.addConnectionNotificationListener(this, null, null);
+        ssProxy.addNotificationListener(this, null, null);
+        this.cmd = repairCmd.start();
+    }
+
+    public void close()
+    {
+        try
+        {
+            ssProxy.removeNotificationListener(this);
+        }
+        catch (ListenerNotFoundException e)
+        {
+            // noop - there may be double removes with error handling
+        }
+        if (jmxc != null)
+        {
+            try
+            {
+                jmxc.removeConnectionNotificationListener(this);
+            }
+            catch (ListenerNotFoundException e)
+            {
+                // noop - there may be double removes with error handling
+            }
+        }
     }
 
     public void run() throws Exception
     {
+        if (cmd == null)
+            return;
+
         if (cmd <= 0)
         {
             // repairAsync can only return 0 for replication factor 1.
-            String message = String.format("Replication factor is 1. No repair is needed for keyspace '%s'", keyspace);
+            String message = String.format("Replication factor is 1. No repair is needed for keyspace '%s'", repairCmd.keyspace);
             printMessage(message);
         }
         else
@@ -117,7 +168,7 @@ public class RepairRunner extends JMXNotificationProgressListener
     {
         error = new IOException(String.format("[%s] JMX connection closed. You should check server log for repair status of keyspace %s"
                                               + "(Subsequent keyspaces are not going to be repaired).",
-                                              format.format(timestamp), keyspace));
+                                              format.format(timestamp), repairCmd.keyspace));
         condition.signalAll();
     }
 

src/java/org/apache/cassandra/tools/nodetool/ConsensusMigrationAdmin.java

@@ -27,8 +27,10 @@ import java.util.Set;
 import io.airlift.airline.Arguments;
 import io.airlift.airline.Command;
 import io.airlift.airline.Option;
+import org.apache.cassandra.service.consensus.migration.ConsensusMigrationTarget;
 import org.apache.cassandra.tools.NodeProbe;
 import org.apache.cassandra.tools.NodeTool;
+import org.apache.cassandra.tools.RepairRunner.RepairCmd;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static java.util.Collections.singleton;
@@ -96,26 +98,50 @@ public abstract class ConsensusMigrationAdmin extends NodeTool.NodeToolCmd
         @Arguments(usage = "[<keyspace> <tables>...]", description = "The keyspace followed by one or many tables")
         private List<String> schemaArgs = new ArrayList<>();
 
+        private static class FinishMigrationRepairCommand extends RepairCmd
+        {
+            private final NodeProbe probe;
+            private final String keyspace;
+            private final List<String> maybeTableNames;
+            private final String maybeRangesStr;
+            private final ConsensusMigrationTarget target;
+
+            public FinishMigrationRepairCommand(NodeProbe probe, String keyspace, List<String> maybeTableNames, String maybeRangesStr, ConsensusMigrationTarget target)
+            {
+                super(keyspace);
+                this.probe = probe;
+                this.keyspace = keyspace;
+                this.maybeTableNames = maybeTableNames;
+                this.maybeRangesStr = maybeRangesStr;
+                this.target = target;
+            }
+
+            @Override
+            public Integer start()
+            {
+                return probe.getStorageService().finishConsensusMigration(keyspace, maybeTableNames, maybeRangesStr, target);
+            }
+        }
+
         protected void execute(NodeProbe probe)
         {
             checkArgument((endToken != null) == (startToken != null), "Start and end token must be specified together");
             String maybeRangesStr = startToken != null ? startToken + ":" + endToken : null;
             List<String> keyspaceNames = parseOptionalKeyspace(schemaArgs, probe, KeyspaceSet.ACCORD_MANAGED);
             List<String> maybeTableNames = schemaArgs.size() > 1 ? schemaArgs.subList(1, schemaArgs.size()) : null;
+            List<RepairCmd> repairCmds = new ArrayList<>(keyspaceNames.size() * 2);
             for (String keyspace : keyspaceNames)
             {
-                List<Integer> commands = probe.getStorageService().finishConsensusMigration(keyspace, maybeTableNames, maybeRangesStr);
+                repairCmds.add(new FinishMigrationRepairCommand(probe, keyspace, maybeTableNames, maybeRangesStr, ConsensusMigrationTarget.paxos));
-                for (Integer command : commands)
+                repairCmds.add(new FinishMigrationRepairCommand(probe, keyspace, maybeTableNames, maybeRangesStr, ConsensusMigrationTarget.accord));
-                {
+            }
-                    try
+            try
-                    {
+            {
-                        probe.blockOnAsyncRepair(probe.output().out, keyspace, command);
+                probe.startAndBlockOnAsyncRepairs(probe.output().out, repairCmds);
-                    }
+            }
-                    catch (IOException e)
+            catch (IOException e)
-                    {
+            {
-                        throw new RuntimeException("Error occurred attempting to finish migration for keyspace " + keyspace + " tables " + maybeTableNames + " and ranges " + maybeRangesStr, e);
+                throw new RuntimeException("Error occurred attempting to finish migration for keyspace(s) " + keyspaceNames + " tables " + maybeTableNames + " and ranges " + maybeRangesStr, e);
-                    }
-                }
             }
             probe.output().out.printf("Finished consensus migration range (%s) of keyspaces %s and tables %s%n", maybeRangesStr, keyspaceNames, maybeTableNames);
         }