natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Allow threads waiting for the log follower to be interrupted 

Patch by Sam Tunnicliffe and David Capwell; reviewed by Alex Petrov for
CASSANDRA-19761
This commit is contained in:
Sam Tunnicliffe 2024-07-11 19:40:55 +01:00
parent 42d18a83e4
commit dc45bb5876
5 changed files with 125 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
5.1
* Allow threads waiting for the metadata log follower to be interrupted (CASSANDRA-19761)
* Support dictionary lookup for CassandraPasswordValidator (CASSANDRA-19762)
* Disallow denylisting keys in system_cluster_metadata (CASSANDRA-19713)
* Fix gossip status after replacement (CASSANDRA-19712)

4
src/java/org/apache/cassandra/tcm/RemoteProcessor.java
View File

@ -152,7 +152,7 @@ public final class RemoteProcessor implements Processor
{
try
{
return fetchLogAndWaitInternal(candidateIterator, log).awaitUninterruptibly().get();
return fetchLogAndWaitInternal(candidateIterator, log).await().get();
}
catch (InterruptedException | ExecutionException e)
{
@ -191,7 +191,7 @@ public final class RemoteProcessor implements Processor
{
Promise<RSP> promise = new AsyncPromise<>();
sendWithCallbackAsync(promise, verb, request, candidates, retryPolicy);
return promise.awaitUninterruptibly().get();
return promise.await().get();
}
catch (InterruptedException | ExecutionException e)
{

22
src/java/org/apache/cassandra/tcm/log/LocalLog.java
View File

@ -370,6 +370,8 @@ public abstract class LocalLog implements Closeable
*/
public void append(LogState logState)
{
if (logState.isEmpty())
return;
logger.debug("Appending log state with snapshot to the pending buffer: {}", logState);
// If we receive a base state (snapshot), we need to construct a synthetic ForceSnapshot transformation that will serve as
// a base for application of the rest of the entries. If the log state contains any additional transformations that follow
@ -403,13 +405,14 @@ public abstract class LocalLog implements Closeable
{
runOnce(null);
}
catch (InterruptedException | TimeoutException e)
catch (TimeoutException e)
{
throw new RuntimeException("Should not have happened, since we await uninterruptibly", e);
// This should not happen as no duration was specified in the call to runOnce
throw new RuntimeException("Timed out waiting for log follower to run", e);
}
}
abstract void runOnce(DurationSpec durationSpec) throws InterruptedException, TimeoutException;
abstract void runOnce(DurationSpec durationSpec) throws TimeoutException;
abstract void processPending();
private Entry peek()
@ -664,8 +667,11 @@ public abstract class LocalLog implements Closeable
}
@Override
public void runOnce(DurationSpec duration) throws InterruptedException, TimeoutException
public void runOnce(DurationSpec duration) throws TimeoutException
{
if (executor.isTerminated())
throw new IllegalStateException("Global log follower has shutdown");
Condition ours = Condition.newOneTimeCondition();
for (int i = 0; i < 2; i++)
{
@ -678,9 +684,10 @@ public abstract class LocalLog implements Closeable
{
if (duration == null)
{
current.awaitUninterruptibly();
current.awaitThrowUncheckedOnInterrupt();
}
else if (!current.await(duration.to(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS))
else if (!current.awaitThrowUncheckedOnInterrupt(duration.to(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS))
{
throw new TimeoutException(String.format("Timed out waiting for follower to run at least once. " +
"Pending is %s and current is now at epoch %s.",
@ -707,7 +714,7 @@ public abstract class LocalLog implements Closeable
if (runnable.subscriber.compareAndSet(null, ours))
{
runnable.logNotifier.signalAll();
ours.awaitUninterruptibly();
ours.awaitThrowUncheckedOnInterrupt();
return;
}
}
@ -727,6 +734,7 @@ public abstract class LocalLog implements Closeable
Condition condition = runnable.subscriber.get();
if (condition != null)
condition.signalAll();
runnable.logNotifier.signalAll();
try
{

3
src/java/org/apache/cassandra/utils/JVMStabilityInspector.java
View File

@ -141,6 +141,9 @@ public final class JVMStabilityInspector
if (t instanceof InterruptedException)
throw new UncheckedInterruptedException((InterruptedException) t);
if (t instanceof UncheckedInterruptedException)
throw (UncheckedInterruptedException)t;
if (DatabaseDescriptor.getDiskFailurePolicy() == Config.DiskFailurePolicy.die)
if (t instanceof FSError || t instanceof CorruptSSTableException)
isUnstable = true;

104
test/distributed/org/apache/cassandra/distributed/test/tcm/CMSShutdownTest.java Normal file
View File

@ -0,0 +1,104 @@
/*
* 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.distributed.test.tcm;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import org.junit.Test;
import net.bytebuddy.ByteBuddy;
import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
import net.bytebuddy.implementation.MethodDelegation;
import net.bytebuddy.implementation.bind.annotation.SuperCall;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.PaxosBackedProcessor;
import org.apache.cassandra.tcm.Transformation;
import org.apache.cassandra.tcm.log.Entry;
import org.apache.cassandra.tcm.transformations.TriggerSnapshot;
import org.apache.cassandra.utils.Shared;
import static net.bytebuddy.matcher.ElementMatchers.named;
public class CMSShutdownTest extends TestBaseImpl
{
@Test
public void shutdownCMSCoincidingWithUnsuccessfulCommit() throws Exception
{
// This test simulates a CMS node attempting to commit an entry to the log but being unable
// to obtain consensus from other CMS members while it is also shutting down itself.
try (Cluster cluster = Cluster.build(2)
.withConfig(c -> c.with(Feature.values()))
.withInstanceInitializer(BBHelper::install)
.start())
{
cluster.get(1).runOnInstance(CommitHelper::scheduleCommits);
State.latch.await();
}
}
private static class CommitHelper
{
public static void commitTransformations()
{
// Continuously attempt to commit a log entry, meanwhile counting down the
// latch ensures that every commit will fail as if unable to obtain consensus
// from other CMS members
State.latch.countDown();
for (; ;)
ClusterMetadataService.instance().commit(TriggerSnapshot.instance);
}
public static void scheduleCommits()
{
Stage.MISC.execute(CommitHelper::commitTransformations);
}
}
@Shared
public static class State
{
public static final CountDownLatch latch = new CountDownLatch(1);
}
public static class BBHelper
{
static void install(ClassLoader cl, int node)
{
if (node != 1) return;
new ByteBuddy().rebase(PaxosBackedProcessor.class)
.method(named("tryCommitOne"))
.intercept(MethodDelegation.to(BBHelper.class))
.make()
.load(cl, ClassLoadingStrategy.Default.INJECTION);
}
public static boolean tryCommitOne(Entry.Id entryId, Transformation transform, Epoch previousEpoch, Epoch nextEpoch, @SuperCall Callable<Boolean> call) throws Exception
{
if (State.latch.getCount() == 1)
return call.call();
return false;
}
}
}