From 0f51ee407bbcc710f2619c8f36bdd1c12329e619 Mon Sep 17 00:00:00 2001 From: Blake Eggleston Date: Tue, 13 Aug 2024 15:24:10 -0700 Subject: [PATCH] CEP-15: C* - Early repair failures hang nodetool Patch by Blake Eggleston; Reviewed by Ariel Weisberg for CASSANDRA-19834 --- .../cassandra/service/StorageService.java | 10 +-- .../service/StorageServiceMBean.java | 8 ++- .../migration/ConsensusTableMigration.java | 28 +++++---- .../org/apache/cassandra/tools/NodeProbe.java | 32 ++++++---- .../apache/cassandra/tools/RepairRunner.java | 63 +++++++++++++++++-- .../nodetool/ConsensusMigrationAdmin.java | 50 +++++++++++---- 6 files changed, 142 insertions(+), 49 deletions(-) diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index ef4068e992..b5de6eae42 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/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 finishConsensusMigration(@Nonnull String keyspace, - @Nullable List maybeTableNames, - @Nullable String maybeRangesStr) + public Integer finishConsensusMigration(@Nonnull String keyspace, + @Nullable List maybeTableNames, + @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 diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index c58205898d..d95ea0f779 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/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 maybeTableNames, @Nullable String maybeRangesStr); - List finishConsensusMigration(@Nonnull String keyspace, - @Nullable List maybeTableNames, - @Nullable String maybeRangesStr); + Integer finishConsensusMigration(@Nonnull String keyspace, + @Nullable List maybeTableNames, + @Nullable String maybeRangesStr, + @Nonnull ConsensusMigrationTarget target); String listConsensusMigrations(@Nullable Set keyspaceNames, @Nullable Set tableNames, @Nonnull String format); diff --git a/src/java/org/apache/cassandra/service/consensus/migration/ConsensusTableMigration.java b/src/java/org/apache/cassandra/service/consensus/migration/ConsensusTableMigration.java index 62d449d5d6..05e351f13d 100644 --- a/src/java/org/apache/cassandra/service/consensus/migration/ConsensusTableMigration.java +++ b/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 finishMigrationToConsensusProtocol(@Nonnull String keyspace, + public static Integer finishMigrationToConsensusProtocol(@Nonnull String keyspace, @Nonnull Optional> maybeTables, - @Nonnull Optional maybeRangesStr) + @Nonnull Optional maybeRangesStr, + ConsensusMigrationTarget target) { checkArgument(!maybeTables.isPresent() || !maybeTables.get().isEmpty(), "Must provide at least 1 table if Optional is not empty"); + checkNotNull(target); Optional>> localKeyspaceRanges = Optional.of(ImmutableList.copyOf(StorageService.instance.getLocalReplicas(keyspace).onlyFull().ranges())); List> ranges = maybeRangesToRanges(maybeRangesStr, localKeyspaceRanges); @@ -215,17 +217,17 @@ public abstract class ConsensusTableMigration tableMigrationStates.add(tms); }); - List migratingToAccord = tableMigrationStates.stream().filter(tms -> tms.targetProtocol == ConsensusMigrationTarget.accord).collect(toImmutableList()); - List migratingToPaxos = tableMigrationStates.stream().filter(tms -> tms.targetProtocol == ConsensusMigrationTarget.paxos).collect(toImmutableList());; - - Integer accordRepairCmd = finishMigrationToAccord(keyspace, migratingToAccord, ranges); - Integer paxosRepairCmd = finishMigrationToPaxos(keyspace, migratingToPaxos, ranges); - List result = new ArrayList<>(); - if (accordRepairCmd != null) - result.add(accordRepairCmd); - if (paxosRepairCmd != null) - result.add(paxosRepairCmd); - return result; + switch (target) + { + case accord: + List migratingToAccord = tableMigrationStates.stream().filter(tms -> tms.targetProtocol == ConsensusMigrationTarget.accord).collect(toImmutableList()); + return finishMigrationToAccord(keyspace, migratingToAccord, ranges); + case paxos: + List migratingToPaxos = tableMigrationStates.stream().filter(tms -> tms.targetProtocol == ConsensusMigrationTarget.paxos).collect(toImmutableList());; + return finishMigrationToPaxos(keyspace, migratingToPaxos, ranges); + default: + throw new IllegalArgumentException("Unsupported target: " + target); + } } private interface MigrationFinisher diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java index e733667b1e..37b7ce627a 100644 --- a/src/java/org/apache/cassandra/tools/NodeProbe.java +++ b/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 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 cmds) throws IOException { - RepairRunner runner = new RepairRunner(out, ssProxy, keyspace, cmd); + List runners = new ArrayList<>(cmds.size()); + for (RepairCmd cmd : cmds) + runners.add(new RepairRunner(out, jmxc, ssProxy, cmd)); + try { - if (jmxc != null) - jmxc.addConnectionNotificationListener(runner, null, null); - ssProxy.addNotificationListener(runner, null, null); - runner.run(); + runners.forEach(RepairRunner::start); + + for (RepairRunner runner : runners) + runner.run(); } catch (Exception e) { @@ -546,9 +558,7 @@ public class NodeProbe implements AutoCloseable { try { - ssProxy.removeNotificationListener(runner); - if (jmxc != null) - jmxc.removeConnectionNotificationListener(runner); + runners.forEach(RepairRunner::close); } catch (Throwable e) { diff --git a/src/java/org/apache/cassandra/tools/RepairRunner.java b/src/java/org/apache/cassandra/tools/RepairRunner.java index 3a4f77ccd2..8451de2918 100644 --- a/src/java/org/apache/cassandra/tools/RepairRunner.java +++ b/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.cmd = cmd; + this.repairCmd = repairCmd; + } + + 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(); } diff --git a/src/java/org/apache/cassandra/tools/nodetool/ConsensusMigrationAdmin.java b/src/java/org/apache/cassandra/tools/nodetool/ConsensusMigrationAdmin.java index 9878034f88..e7e1fd9cfc 100644 --- a/src/java/org/apache/cassandra/tools/nodetool/ConsensusMigrationAdmin.java +++ b/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 = "[ ...]", description = "The keyspace followed by one or many tables") private List schemaArgs = new ArrayList<>(); + private static class FinishMigrationRepairCommand extends RepairCmd + { + private final NodeProbe probe; + private final String keyspace; + private final List maybeTableNames; + private final String maybeRangesStr; + private final ConsensusMigrationTarget target; + + public FinishMigrationRepairCommand(NodeProbe probe, String keyspace, List 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 keyspaceNames = parseOptionalKeyspace(schemaArgs, probe, KeyspaceSet.ACCORD_MANAGED); List maybeTableNames = schemaArgs.size() > 1 ? schemaArgs.subList(1, schemaArgs.size()) : null; + List repairCmds = new ArrayList<>(keyspaceNames.size() * 2); for (String keyspace : keyspaceNames) { - List commands = probe.getStorageService().finishConsensusMigration(keyspace, maybeTableNames, maybeRangesStr); - for (Integer command : commands) - { - try - { - probe.blockOnAsyncRepair(probe.output().out, keyspace, command); - } - catch (IOException e) - { - throw new RuntimeException("Error occurred attempting to finish migration for keyspace " + keyspace + " tables " + maybeTableNames + " and ranges " + maybeRangesStr, e); - } - } + repairCmds.add(new FinishMigrationRepairCommand(probe, keyspace, maybeTableNames, maybeRangesStr, ConsensusMigrationTarget.paxos)); + repairCmds.add(new FinishMigrationRepairCommand(probe, keyspace, maybeTableNames, maybeRangesStr, ConsensusMigrationTarget.accord)); + } + try + { + probe.startAndBlockOnAsyncRepairs(probe.output().out, repairCmds); + } + catch (IOException 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); }