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

Accord simulation test failing with "ClassNotFoundException: WARN [AccordExecutor[1,0]" 

patch by David Capwell; reviewed by Benedict Elliott Smith for CASSANDRA-20127
This commit is contained in:
David Capwell 2024-12-03 20:23:38 -08:00
parent a5dc44ad57
commit c114fc46d2
10 changed files with 164 additions and 104 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -115,8 +115,8 @@ public class FetchCMSLog
// If both we and the other node believe it should be caught up with a linearizable read
boolean consistentFetch = request.consistentFetch && !ClusterMetadataService.instance().isCurrentMember(message.from());
Retry.Deadline retry = Retry.Deadline.retryIndefinitely(DatabaseDescriptor.getCmsAwaitTimeout().to(TimeUnit.NANOSECONDS),
TCMMetrics.instance.fetchLogRetries);
Retry.Deadline retry = Retry.Deadline.after(DatabaseDescriptor.getCmsAwaitTimeout().to(TimeUnit.NANOSECONDS),
new Retry.Jitter(TCMMetrics.instance.fetchLogRetries));
LogState delta;
if (consistentFetch)
delta = processor.get().getLogState(message.payload.lowerBound, Epoch.MAX, false, retry);

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

@ -97,8 +97,8 @@ public class ReconstructLogState
throw new NotCMSException("This node is not in the CMS, can't generate a consistent log fetch response to " + message.from());
LogState result = processor.get().getLogState(request.lowerBound, request.higherBound, request.includeSnapshot,
Retry.Deadline.retryIndefinitely(DatabaseDescriptor.getCmsAwaitTimeout().to(TimeUnit.NANOSECONDS),
TCMMetrics.instance.fetchLogRetries));
Retry.Deadline.after(DatabaseDescriptor.getCmsAwaitTimeout().to(TimeUnit.NANOSECONDS),
new Retry.Jitter(TCMMetrics.instance.fetchLogRetries)));
MessagingService.instance().send(message.responseWith(result), message.from());
}

29
src/java/org/apache/cassandra/tcm/Retry.java
View File

@ -208,35 +208,6 @@ public abstract class Retry
return new Deadline(Clock.Global.nanoTime() + timeoutNanos, delegate);
}
/**
* Since we are using message expiration for communicating timeouts to CMS nodes, we have to be careful not
* to overflow the long, since messaging is using only 32 bits for deadlines. To achieve that, we are
* giving `timeoutNanos` every time we retry, but will retry indefinitely.
*/
public static Deadline retryIndefinitely(long timeoutNanos, Meter retryMeter)
{
return new Deadline(Clock.Global.nanoTime() + timeoutNanos,
new Retry.Jitter(Integer.MAX_VALUE, DEFAULT_BACKOFF_MS, new Random(), retryMeter))
{
@Override
public boolean reachedMax()
{
return false;
}
@Override
public long remainingNanos()
{
return timeoutNanos;
}
public String toString()
{
return String.format("RetryIndefinitely{tries=%d}", currentTries());
}
};
}
public static Deadline wrap(Retry delegate)
{
long deadlineMillis = delegate.maxTries * delegate.maxWait();

92
test/simulator/main/org/apache/cassandra/simulator/paxos/AbstractPairOfSequencesPaxosSimulation.java
View File

@ -29,6 +29,7 @@ import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -130,53 +131,60 @@ abstract class AbstractPairOfSequencesPaxosSimulation extends PaxosSimulation
{
LogAction logs = inst.logs();
LogResult<List<String>> errors = logs.grepForErrors();
if (!errors.getResult().isEmpty())
{
List<Pair<String, String>> errorsSeen = new ArrayList<>();
for (String error : errors.getResult())
{
for (String line : error.split("\\n"))
{
line = line.trim();
if (line.startsWith("ERROR")) continue;
if (line.startsWith("at ")) continue;
errorsSeen.add(Pair.create(line.split(":")[0], error));
break;
}
}
Class<? extends Throwable>[] expected = expectedExceptions();
StringBuilder sb = new StringBuilder();
for (Pair<String, String> pair : errorsSeen)
{
String name = pair.left;
String exception = pair.right;
Class<?> klass;
try
{
klass = Class.forName(name);
}
catch (ClassNotFoundException e)
{
throw new RuntimeException(e);
}
if (!Stream.of(expected).anyMatch(e -> e.isAssignableFrom(klass)))
sb.append("Unexpected exception:\n").append(exception).append('\n');
}
if (sb.length() > 0)
{
AssertionError error = new AssertionError("Saw errors in node" + inst.config().num() + ": " + sb);
// this stacktrace isn't helpful, can be more confusing
error.setStackTrace(new StackTraceElement[0]);
throw error;
}
}
checkErrorLogs(inst.config().num(), logs.grepForErrors());
return ActionList.empty();
}
};
}
@VisibleForTesting
protected void checkErrorLogs(int node, LogResult<List<String>> errors)
{
if (!errors.getResult().isEmpty())
{
List<Pair<String, String>> errorsSeen = new ArrayList<>();
for (String error : errors.getResult())
{
for (String line : error.split("\\n"))
{
line = line.trim();
if (line.startsWith("ERROR")) continue;
if (line.startsWith("WARN")) continue;
if (line.startsWith("at ")) continue;
errorsSeen.add(Pair.create(line.split(":")[0], error));
break;
}
}
Class<? extends Throwable>[] expected = expectedExceptions();
StringBuilder sb = new StringBuilder();
for (Pair<String, String> pair : errorsSeen)
{
String name = pair.left;
String exception = pair.right;
Class<?> klass;
try
{
klass = Class.forName(name);
}
catch (ClassNotFoundException e)
{
sb.append("Unexpected exception (could not parse line):\n").append(exception).append('\n');
continue;
}
if (!Stream.of(expected).anyMatch(e -> e.isAssignableFrom(klass)))
sb.append("Unexpected exception:\n").append(exception).append('\n');
}
if (sb.length() > 0)
{
AssertionError error = new AssertionError("Saw errors in node" + node + ": " + sb);
// this stacktrace isn't helpful, can be more confusing
error.setStackTrace(new StackTraceElement[0]);
throw error;
}
}
}
protected Metrics getMetrics(int coordinatorIndex)
{
return cluster.get(coordinatorIndex).metrics();

23
test/simulator/main/org/apache/cassandra/simulator/paxos/AccordClusterSimulation.java
View File

@ -22,7 +22,6 @@ import java.io.IOException;
import org.apache.cassandra.simulator.ClusterSimulation;
import org.apache.cassandra.simulator.RandomSource;
import org.apache.cassandra.simulator.utils.IntRange;
import org.apache.cassandra.simulator.utils.KindOfSequence;
import static java.util.concurrent.TimeUnit.SECONDS;
@ -39,28 +38,6 @@ class AccordClusterSimulation extends ClusterSimulation<PaxosSimulation> impleme
random.reset(seed);
return new AccordClusterSimulation(random, seed, uniqueNum, this);
}
public void applyHandicaps()
{
/**
* TODO (required): remove
* We currently require coordinators to have a CommandStore to coordinate a query, but not every node
* is a replica under standard simulation
*
* The current homekey implementation isn't compatible with the C* commands per key implementation when
* a non-replica coordinates a query.
*
* This creates a few problems.
*
* First when a non-replica coordinator chooses a home key, it chooses the end of one of it's ranges and
* adds it to the txn. This doesn't work with the C* CFK implementation, because it expects a partition
* key. This will change with the partial replication patch, so we can re-evaluate then.
*
* Second, nodes that haven't joined the ring have no ranges to pull home keys from, so they npe
*/
dcCount = new IntRange(1, 1);
nodeCount = new IntRange(3, 3);
}
}
AccordClusterSimulation(RandomSource random, long seed, int uniqueNum, Builder builder) throws IOException

3
test/simulator/main/org/apache/cassandra/simulator/paxos/AccordSimulationRunner.java
View File

@ -46,7 +46,6 @@ public class AccordSimulationRunner extends SimulationRunner
protected void run(long seed, AccordClusterSimulation.Builder builder) throws IOException
{
beforeAll();
builder.applyHandicaps();
super.run(seed, builder);
}
}
@ -60,7 +59,6 @@ public class AccordSimulationRunner extends SimulationRunner
protected void run(long seed, AccordClusterSimulation.Builder builder) throws IOException
{
beforeAll();
builder.applyHandicaps();
super.run(seed, builder);
}
}
@ -74,7 +72,6 @@ public class AccordSimulationRunner extends SimulationRunner
protected void run(long seed, AccordClusterSimulation.Builder builder) throws IOException
{
beforeAll();
builder.applyHandicaps();
super.run(seed, builder);
}
}

11
test/simulator/main/org/apache/cassandra/simulator/paxos/PairOfSequencesAccordSimulation.java
View File

@ -33,6 +33,7 @@ import javax.annotation.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.coordinate.CoordinationFailed;
import com.carrotsearch.hppc.IntArrayList;
import com.carrotsearch.hppc.IntHashSet;
import com.carrotsearch.hppc.cursors.IntCursor;
@ -48,6 +49,7 @@ import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.api.QueryResults;
import org.apache.cassandra.distributed.api.SimpleQueryResult;
import org.apache.cassandra.distributed.impl.Query;
import org.apache.cassandra.exceptions.RequestExecutionException;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.consensus.TransactionalMode;
@ -150,6 +152,13 @@ public class PairOfSequencesAccordSimulation extends AbstractPairOfSequencesPaxo
this.validator = validator;
}
@Override
protected Class<? extends Throwable>[] expectedExceptions()
{
return (Class<? extends Throwable>[]) new Class<?>[] { RequestExecutionException.class,
CoordinationFailed.class };
}
@Override
protected String createTableStmt()
{
@ -182,7 +191,7 @@ public class PairOfSequencesAccordSimulation extends AbstractPairOfSequencesPaxo
public ReadWriteOperation(int id, int[] primaryKeys, IntHashSet reads, IntHashSet writes, IInvokableInstance instance)
{
super(primaryKeys, id, instance, "Accord", createQuery(id, reads, writes));
super(primaryKeys, id, instance, "Accord ReadWrite Txn", createQuery(id, reads, writes));
this.reads = reads;
this.writes = writes;
}

2
test/simulator/main/org/apache/cassandra/simulator/paxos/PaxosSimulation.java
View File

@ -72,7 +72,7 @@ public abstract class PaxosSimulation implements Simulation, ClusterActionListen
private static String createDescription(int[] primaryKeys, int id, String idString)
{
return primaryKeys.length == 1 ? Integer.toString(primaryKeys[0]) : Arrays.toString(primaryKeys) + "/" + id + ": " + idString;
return (primaryKeys.length == 1 ? Integer.toString(primaryKeys[0]) : Arrays.toString(primaryKeys)) + "/" + id + ": " + idString;
}
protected Class<? extends Throwable>[] expectedExceptions()

2
test/simulator/main/org/apache/cassandra/simulator/paxos/PaxosTopologyChangeVerifier.java
View File

@ -58,7 +58,7 @@ public class PaxosTopologyChangeVerifier implements TopologyChangeValidator
@Override
public void after(Topology topologyAfter)
{
afterInternal(topologyAfter.select(topologyBefore.primaryKeys));
afterInternal(topologyBefore == null ? topologyAfter : topologyAfter.select(topologyBefore.primaryKeys));
}
public void afterInternal(Topology topologyAfter)

98
test/simulator/test/org/apache/cassandra/simulator/paxos/AbstractPairOfSequencesPaxosSimulationTest.java Normal file
View File

@ -0,0 +1,98 @@
/*
* 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.simulator.paxos;
import java.util.Collections;
import java.util.List;
import org.junit.Test;
import org.apache.cassandra.distributed.api.LogAction;
import org.apache.cassandra.distributed.api.LogResult;
import org.assertj.core.api.Assertions;
import org.mockito.Mockito;
public class AbstractPairOfSequencesPaxosSimulationTest
{
@Test
public void parseSuccess()
{
String log = "WARN [AccordExecutor[1,0]:1] node1 2024-12-03 17:45:24,574 [10,1577851211987004,9(RX),1]: Exception coordinating ExclusiveSyncPoint for [1b255f4d-ef25-40a6-0000-000000000009:[(-2978380553567688022,-2930342157542402732]]] durability. Increased numberOfSplits to 256\n" +
"accord.coordinate.Invalidated: null\n" +
"\tat accord.coordinate.Propose$Invalidate.lambda$proposeAndCommitInvalidate$2(Propose.java:193)\n" +
"\tat accord.local.Node.withEpoch(Node.java:391)\n" +
"\tat accord.coordinate.Propose$Invalidate.lambda$proposeAndCommitInvalidate$3(Propose.java:186)\n" +
"\tat accord.coordinate.Propose$Invalidate.onSuccess(Propose.java:217)\n" +
"\tat accord.coordinate.Propose$Invalidate.onSuccess(Propose.java:146)\n" +
"\tat accord.impl.RequestCallbacks$CallbackStripe$RegisteredCallback.unsafeOnSuccess(RequestCallbacks.java:119)\n" +
"\tat accord.impl.RequestCallbacks$CallbackStripe.lambda$onSuccess$0(RequestCallbacks.java:189)\n" +
"\tat accord.impl.RequestCallbacks$CallbackStripe$RegisteredCallback.lambda$safeInvoke$0(RequestCallbacks.java:140)\n" +
"\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)\n" +
"\tat accord.utils.async.AsyncChains.lambda$encapsulate$0(AsyncChains.java:498)\n" +
"\tat org.apache.cassandra.service.accord.AccordExecutor$PlainRunnable.run(AccordExecutor.java:989)\n" +
"\tat org.apache.cassandra.service.accord.AccordExecutor$CommandStoreQueue.run(AccordExecutor.java:729)\n" +
"\tat org.apache.cassandra.service.accord.AccordExecutorSimple.run(AccordExecutorSimple.java:95)\n" +
"\tat org.apache.cassandra.concurrent.FutureTask$2.call(FutureTask.java:124)\n" +
"\tat org.apache.cassandra.concurrent.SyncFutureTask.run(SyncFutureTask.java:68)\n" +
"\tat org.apache.cassandra.simulator.systems.InterceptingExecutor$AbstractSingleThreadedExecutorPlus.lambda$new$0(InterceptingExecutor.java:585)\n" +
"\tat io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)\n" +
"\tat java.base/java.lang.Thread.run(Thread.java:829)";
LogResult<List<String>> errors = new LogAction.BasicLogResult<>(42, Collections.singletonList(log));
AbstractPairOfSequencesPaxosSimulation simulation = Mockito.mock(AbstractPairOfSequencesPaxosSimulation.class);
Mockito.doCallRealMethod().when(simulation).checkErrorLogs(Mockito.eq(0), Mockito.eq(errors));
Mockito.when(simulation.expectedExceptions()).thenReturn((Class<? extends Throwable>[]) new Class<?>[] { accord.coordinate.Invalidated.class });
simulation.checkErrorLogs(0, errors);
}
@Test
public void parseFailure()
{
String log = "FAKE [AccordExecutor[1,0]:1] node1 2024-12-03 17:45:24,574 [10,1577851211987004,9(RX),1]: Exception coordinating ExclusiveSyncPoint for [1b255f4d-ef25-40a6-0000-000000000009:[(-2978380553567688022,-2930342157542402732]]] durability. Increased numberOfSplits to 256\n" +
"accord.coordinate.Invalidated: null\n" +
"\tat accord.coordinate.Propose$Invalidate.lambda$proposeAndCommitInvalidate$2(Propose.java:193)\n" +
"\tat accord.local.Node.withEpoch(Node.java:391)\n" +
"\tat accord.coordinate.Propose$Invalidate.lambda$proposeAndCommitInvalidate$3(Propose.java:186)\n" +
"\tat accord.coordinate.Propose$Invalidate.onSuccess(Propose.java:217)\n" +
"\tat accord.coordinate.Propose$Invalidate.onSuccess(Propose.java:146)\n" +
"\tat accord.impl.RequestCallbacks$CallbackStripe$RegisteredCallback.unsafeOnSuccess(RequestCallbacks.java:119)\n" +
"\tat accord.impl.RequestCallbacks$CallbackStripe.lambda$onSuccess$0(RequestCallbacks.java:189)\n" +
"\tat accord.impl.RequestCallbacks$CallbackStripe$RegisteredCallback.lambda$safeInvoke$0(RequestCallbacks.java:140)\n" +
"\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)\n" +
"\tat accord.utils.async.AsyncChains.lambda$encapsulate$0(AsyncChains.java:498)\n" +
"\tat org.apache.cassandra.service.accord.AccordExecutor$PlainRunnable.run(AccordExecutor.java:989)\n" +
"\tat org.apache.cassandra.service.accord.AccordExecutor$CommandStoreQueue.run(AccordExecutor.java:729)\n" +
"\tat org.apache.cassandra.service.accord.AccordExecutorSimple.run(AccordExecutorSimple.java:95)\n" +
"\tat org.apache.cassandra.concurrent.FutureTask$2.call(FutureTask.java:124)\n" +
"\tat org.apache.cassandra.concurrent.SyncFutureTask.run(SyncFutureTask.java:68)\n" +
"\tat org.apache.cassandra.simulator.systems.InterceptingExecutor$AbstractSingleThreadedExecutorPlus.lambda$new$0(InterceptingExecutor.java:585)\n" +
"\tat io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)\n" +
"\tat java.base/java.lang.Thread.run(Thread.java:829)";
LogResult<List<String>> errors = new LogAction.BasicLogResult<>(42, Collections.singletonList(log));
AbstractPairOfSequencesPaxosSimulation simulation = Mockito.mock(AbstractPairOfSequencesPaxosSimulation.class);
Mockito.doCallRealMethod().when(simulation).checkErrorLogs(Mockito.eq(0), Mockito.eq(errors));
Mockito.when(simulation.expectedExceptions()).thenReturn((Class<? extends Throwable>[]) new Class<?>[] { accord.coordinate.Invalidated.class });
Assertions.assertThatThrownBy(() -> simulation.checkErrorLogs(0, errors))
.isInstanceOf(AssertionError.class)
.hasMessageStartingWith("Saw errors in node0: Unexpected exception (could not parse line):");
}
}