Merge c5ee3a4276 into bdbdf8d710

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

@@ -30,6 +30,7 @@ import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -1247,6 +1248,11 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
         return futures;
     }
 
+    public Future<?> schedulePaxosCleanup(SharedContext ctx, Collection<InetAddressAndPort> endpoints, TableMetadata table, Collection<Range<Token>> ranges, Executor executor)
+    {
+        return PaxosCleanup.cleanup(ctx, endpoints, table, ranges, false, executor);
+    }
+
     public int getPaxosRepairParallelism()
     {
         return DatabaseDescriptor.getPaxosRepairParallelism();

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

@@ -3301,6 +3301,85 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
         logger.info("paxos repair for {} complete", reason);
     }
 
+    public void paxosCleanup(String keyspace, String... tables)
+    {
+        Keyspaces keyspaces = keyspace == null || keyspace.isEmpty()
+                              ? Schema.instance.distributedKeyspaces().without(SchemaConstants.REPLICATED_SYSTEM_KEYSPACE_NAMES)
+                              : Keyspaces.of(Schema.instance.getKeyspaceMetadata(keyspace));
+
+        if (!tablesSpecified(tables) && keyspaces.isEmpty())
+            throw new IllegalArgumentException("No keyspaces available for paxos cleanup");
+
+        for (String ksName : keyspaces.names())
+        {
+            Keyspace ks = Keyspace.open(ksName);
+            Collection<Range<Token>> ranges = getLocalReplicas(ksName).onlyFull().ranges();
+            if (ranges.isEmpty())
+                continue;
+
+            List<TableMetadata> selectedTables = tablesSpecified(tables)
+                                                ? resolveTables(ksName, tables)
+                                                : Lists.newArrayList(ClusterMetadata.current().schema.getKeyspaces().getNullable(ksName).tables);
+
+            for (TableMetadata table : selectedTables)
+            {
+                if (!table.supportsPaxosOperations())
+                    continue;
+
+                logger.info("Starting paxos cleanup for {}.{}", table.keyspace, table.name);
+
+                for (Range<Token> range : ranges)
+                {
+                    EndpointsForRange replicas = ClusterMetadata.current()
+                                                                .placements.get(ks.getMetadata().params.replication)
+                                                                .reads.forRange(range.right).get();
+                    Set<InetAddressAndPort> liveEndpoints = replicas.filter(FailureDetector.isReplicaAlive).endpoints();
+
+                    waitOnPaxosCleanup(liveEndpoints, table, Collections.singleton(range));
+                }
+            }
+        }
+    }
+
+    private static boolean tablesSpecified(String... tables)
+    {
+        return tables != null && tables.length > 0;
+    }
+
+    private static List<TableMetadata> resolveTables(String keyspace, String... tables)
+    {
+        List<TableMetadata> resolved = new ArrayList<>(tables.length);
+        for (String table : tables)
+        {
+            TableMetadata metadata = Schema.instance.getTableMetadata(keyspace, table);
+            if (metadata == null)
+                throw new IllegalArgumentException("Unknown table: " + keyspace + "." + table);
+            resolved.add(metadata);
+        }
+        return resolved;
+    }
+
+    private static void waitOnPaxosCleanup(Collection<InetAddressAndPort> liveEndpoints, TableMetadata table, Collection<Range<Token>> ranges)
+    {
+        try
+        {
+            ActiveRepairService.instance().schedulePaxosCleanup(SharedContext.Global.instance,
+                                                                liveEndpoints,
+                                                                table,
+                                                                ranges,
+                                                                repairCommandExecutor()).get();
+        }
+        catch (InterruptedException e)
+        {
+            Thread.currentThread().interrupt();
+            throw new RuntimeException("Interrupted while running paxos cleanup", e);
+        }
+        catch (ExecutionException e)
+        {
+            throw new RuntimeException("Error during paxos cleanup", e);
+        }
+    }
+
     @VisibleForTesting
     public Future<?> startRepairPaxosForTopologyChange(String reason)
     {

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

@@ -339,6 +339,12 @@ public interface StorageServiceMBean extends NotificationEmitter
     @Deprecated(since = "5.1")
     public void clearSnapshot(Map<String, Object> options, String tag, String... keyspaceNames) throws IOException;
 
+    /**
+     * Clean up paxos state for the supplied keyspace/tables without running full repair.
+     * Blocks until cleanup completes.
+     */
+    void paxosCleanup(String keyspace, String... tables);
+
     /**
      * Get the details of all the snapshot
      * @return A map of snapshotName to all its details in Tabular form.

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

@@ -1052,6 +1052,11 @@ public class NodeProbe implements AutoCloseable
         return snapshotProxy.getTrueSnapshotSize();
     }
 
+    public void paxosCleanup(String keyspace, String... tables) throws IOException
+    {
+        ssProxy.paxosCleanup(keyspace, tables);
+    }
+
     public boolean isJoined()
     {
         return ssProxy.isJoined();

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

@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.tools.nodetool;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.cassandra.tools.NodeProbe;
+import org.apache.cassandra.tools.nodetool.layout.CassandraUsage;
+
+import picocli.CommandLine.Command;
+import picocli.CommandLine.Parameters;
+
+@Command(name = "clearpaxos", description = "Clean up paxos state for one or more tables without running repair")
+public class ClearPaxos extends AbstractCommand
+{
+    @CassandraUsage(usage = "[<keyspace> <tables>...]", description = "The keyspace followed by optional tables to clean up paxos state")
+    @Parameters(description = "The keyspace followed by optional tables to clean up paxos state", arity = "0..*")
+    private List<String> args = new ArrayList<>();
+
+    @Override
+    public void execute(NodeProbe probe)
+    {
+        String keyspace = args.isEmpty() ? null : args.get(0);
+        String[] tables = args.size() <= 1 ? new String[0] : args.subList(1, args.size()).toArray(new String[0]);
+
+        try
+        {
+            probe.output().out.println("Starting paxos cleanup...");
+            probe.paxosCleanup(keyspace, tables);
+            probe.output().out.println("Paxos cleanup completed.");
+        }
+        catch (IOException e)
+        {
+            throw new RuntimeException("Error during paxos cleanup", e);
+        }
+    }
+}

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

@@ -67,7 +67,8 @@ import static org.apache.cassandra.tools.nodetool.Help.printTopCommandUsage;
                          CMSAdmin.class,
                          Cleanup.class,
                          ClearSnapshot.class,
-                         ClientStats.class,
+                          ClearPaxos.class,
+                          ClientStats.class,
                          Compact.class,
                          CompactionHistory.class,
                          CompactionStats.class,

test/distributed/org/apache/cassandra/distributed/test/NodeToolTest.java

@@ -26,10 +26,16 @@ import org.junit.BeforeClass;
 import org.junit.Test;
 
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.marshal.Int32Type;
 import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
 import org.apache.cassandra.distributed.api.ICluster;
 import org.apache.cassandra.distributed.api.IInvokableInstance;
 import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.service.paxos.Ballot;
 
 import static org.junit.Assert.assertEquals;
 
@@ -154,4 +160,45 @@ public class NodeToolTest extends TestBaseImpl
             .success()
             .stdoutContains("GitSHA:");
     }
+
+    @Test
+    public void testClearPaxos() throws Throwable
+    {
+        CLUSTER.schemaChange("CREATE KEYSPACE IF NOT EXISTS clearpaxos_test WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1}");
+        CLUSTER.schemaChange("CREATE TABLE IF NOT EXISTS clearpaxos_test.tbl (id int PRIMARY KEY, v int)");
+
+        long before = NODE.callsOnInstance(() -> {
+            ColumnFamilyStore cfs = Keyspace.open("clearpaxos_test").getColumnFamilyStore("tbl");
+            DecoratedKey key = cfs.decorateKey(Int32Type.instance.decompose(1));
+            return cfs.getPaxosRepairHistory().ballotForToken(key.getToken()).uuidTimestamp();
+        }).call();
+
+        CLUSTER.coordinator(1)
+               .execute("INSERT INTO clearpaxos_test.tbl (id, v) VALUES (1, 1) IF NOT EXISTS", ConsistencyLevel.QUORUM);
+
+        NODE.nodetoolResult("clearpaxos", "clearpaxos_test", "tbl")
+            .asserts()
+            .success()
+            .stdoutContains("Paxos cleanup completed.");
+
+        long after = NODE.callsOnInstance(() -> {
+            ColumnFamilyStore cfs = Keyspace.open("clearpaxos_test").getColumnFamilyStore("tbl");
+            DecoratedKey key = cfs.decorateKey(Int32Type.instance.decompose(1));
+            return cfs.getPaxosRepairHistory().ballotForToken(key.getToken()).uuidTimestamp();
+        }).call();
+
+        assertEquals(Ballot.none().uuidTimestamp(), before);
+        org.junit.Assert.assertTrue("Expected paxos repair history to be updated", after > before);
+    }
+
+    @Test
+    public void testClearPaxosUnknownTable() throws Throwable
+    {
+        CLUSTER.schemaChange("CREATE KEYSPACE IF NOT EXISTS clearpaxos_test WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1}");
+
+        NODE.nodetoolResult("clearpaxos", "clearpaxos_test", "missing_tbl")
+            .asserts()
+            .failure()
+            .stdoutContains("Unknown table: clearpaxos_test.missing_tbl");
+    }
 }

test/unit/org/apache/cassandra/tools/NodeToolCommandTest.java

@@ -84,4 +84,15 @@ public class NodeToolCommandTest
         Assert.assertEquals(options.get(RepairOption.INCREMENTAL_KEY), Boolean.toString(false));
     }
 
+    @Test
+    public void clearPaxosCommandTest() throws IOException
+    {
+        int result = new NodeTool(repairNodeFactory, output).execute("clearpaxos", "ks", "tbl1", "tbl2");
+        Assert.assertEquals(0, result);
+
+        ArgumentCaptor<String[]> tablesCaptor = ArgumentCaptor.forClass(String[].class);
+        verify(nodeProbe).paxosCleanup(Mockito.eq("ks"), tablesCaptor.capture());
+        Assert.assertArrayEquals(new String[]{ "tbl1", "tbl2" }, tablesCaptor.getValue());
+    }
+
 }