You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by bd...@apache.org on 2024/09/17 20:50:43 UTC
(cassandra) branch cep-15-accord updated: CEP-15: C* - Early repair failures hang nodetool
This is an automated email from the ASF dual-hosted git repository.
bdeggleston pushed a commit to branch cep-15-accord
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cep-15-accord by this push:
new ae720a24ee CEP-15: C* - Early repair failures hang nodetool
ae720a24ee is described below
commit ae720a24ee2f0209c10d8eeadd901c014348f471
Author: Blake Eggleston <bl...@ultrablake.com>
AuthorDate: Tue Aug 13 15:24:10 2024 -0700
CEP-15: C* - Early repair failures hang nodetool
Patch by Blake Eggleston; Reviewed by Ariel Weisberg for CASSANDRA-19834
---
.../apache/cassandra/service/StorageService.java | 10 ++--
.../cassandra/service/StorageServiceMBean.java | 8 +--
.../migration/ConsensusTableMigration.java | 28 +++++-----
src/java/org/apache/cassandra/tools/NodeProbe.java | 30 +++++++----
.../org/apache/cassandra/tools/RepairRunner.java | 63 +++++++++++++++++++---
.../tools/nodetool/ConsensusMigrationAdmin.java | 50 ++++++++++++-----
6 files changed, 141 insertions(+), 48 deletions(-)
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index 14f17aff0f..60bea08d4d 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -172,6 +172,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;
@@ -1651,12 +1652,13 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
}
@Override
- public List<Integer> finishConsensusMigration(@Nonnull String keyspace,
- @Nullable List<String> maybeTableNames,
- @Nullable String maybeRangesStr)
+ public Integer finishConsensusMigration(@Nonnull String keyspace,
+ @Nullable List<String> 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 c4072a35c0..de7927bcd3 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
@@ -1077,9 +1078,10 @@ public interface StorageServiceMBean extends NotificationEmitter
@Nullable List<String> maybeTableNames,
@Nullable String maybeRangesStr);
- List<Integer> finishConsensusMigration(@Nonnull String keyspace,
- @Nullable List<String> maybeTableNames,
- @Nullable String maybeRangesStr);
+ Integer finishConsensusMigration(@Nonnull String keyspace,
+ @Nullable List<String> maybeTableNames,
+ @Nullable String maybeRangesStr,
+ @Nonnull ConsensusMigrationTarget target);
String listConsensusMigrations(@Nullable Set<String> keyspaceNames, @Nullable Set<String> 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 00bf2f426f..0190c6c312 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<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());
- List<TableMigrationState> migratingToPaxos = tableMigrationStates.stream().filter(tms -> tms.targetProtocol == ConsensusMigrationTarget.paxos).collect(toImmutableList());;
-
- Integer accordRepairCmd = finishMigrationToAccord(keyspace, migratingToAccord, ranges);
- Integer paxosRepairCmd = finishMigrationToPaxos(keyspace, migratingToPaxos, ranges);
- List<Integer> result = new ArrayList<>();
- if (accordRepairCmd != null)
- result.add(accordRepairCmd);
- if (paxosRepairCmd != null)
- result.add(paxosRepairCmd);
- return result;
+ switch (target)
+ {
+ case accord:
+ List<TableMigrationState> migratingToAccord = tableMigrationStates.stream().filter(tms -> tms.targetProtocol == ConsensusMigrationTarget.accord).collect(toImmutableList());
+ return finishMigrationToAccord(keyspace, migratingToAccord, ranges);
+ case paxos:
+ List<TableMigrationState> 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 aaa4b996b0..59a732b1d1 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;
@@ -121,6 +122,7 @@ 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;
@@ -512,18 +514,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)
- 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)
{
@@ -533,9 +545,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 0208c431fc..21b7a7e727 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.StorageServiceMBean;
import org.apache.cassandra.utils.concurrent.Condition;
import org.apache.cassandra.utils.progress.ProgressEvent;
@@ -41,30 +44,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
@@ -116,7 +167,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 = "[<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);
- 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);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org