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

Accord: Test fixes 

patch by Alex Petrov; reviewed by Benedict Elliott Smith, David Capwell for CASSANDRA-20552
This commit is contained in:
Alex Petrov 2025-04-14 16:24:09 -07:00 committed by David Capwell
parent d3439c671a
commit 4cdbb5cf2a
4 changed files with 22 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/java/org/apache/cassandra/net/Verb.java
View File

@ -293,7 +293,7 @@ public enum Verb
// transactional cluster metadata
TCM_COMMIT_RSP (801, P0, rpcTimeout, INTERNAL_METADATA, MessageSerializers::commitResultSerializer, RESPONSE_HANDLER ),
TCM_COMMIT_REQ (802, P0, rpcTimeout, INTERNAL_METADATA, MessageSerializers::commitSerializer, () -> commitRequestHandler(), TCM_COMMIT_RSP ),
TCM_FETCH_CMS_LOG_RSP (803, P0, rpcTimeout, FETCH_METADATA, MessageSerializers::logStateSerializer, RESPONSE_HANDLER ),
TCM_FETCH_CMS_LOG_RSP (803, P0, shortTimeout, FETCH_METADATA, MessageSerializers::logStateSerializer, RESPONSE_HANDLER ),
TCM_FETCH_CMS_LOG_REQ (804, P0, rpcTimeout, FETCH_METADATA, () -> FetchCMSLog.serializer, () -> fetchLogRequestHandler(), TCM_FETCH_CMS_LOG_RSP ),
TCM_REPLICATION (805, P0, rpcTimeout, INTERNAL_METADATA, MessageSerializers::logStateSerializer, () -> replicationHandler() ),
TCM_NOTIFY_RSP (806, P0, rpcTimeout, INTERNAL_METADATA, () -> Epoch.messageSerializer, RESPONSE_HANDLER ),
@ -304,7 +304,7 @@ public enum Verb
TCM_ABORT_MIG (811, P0, rpcTimeout, INTERNAL_METADATA, () -> CMSInitializationRequest.Initiator.serializer,() -> Election.instance.abortHandler, TCM_INIT_MIG_RSP ),
TCM_DISCOVER_RSP (812, P0, rpcTimeout, INTERNAL_METADATA, () -> Discovery.serializer, RESPONSE_HANDLER ),
TCM_DISCOVER_REQ (813, P0, rpcTimeout, INTERNAL_METADATA, () -> NoPayload.serializer, () -> Discovery.instance.requestHandler, TCM_DISCOVER_RSP ),
TCM_FETCH_PEER_LOG_RSP (818, P0, rpcTimeout, FETCH_METADATA, MessageSerializers::logStateSerializer, RESPONSE_HANDLER ),
TCM_FETCH_PEER_LOG_RSP (818, P0, shortTimeout, FETCH_METADATA, MessageSerializers::logStateSerializer, RESPONSE_HANDLER ),
TCM_FETCH_PEER_LOG_REQ (819, P0, rpcTimeout, FETCH_METADATA, () -> FetchPeerLog.serializer, () -> FetchPeerLog.Handler.instance, TCM_FETCH_PEER_LOG_RSP ),
TCM_RECONSTRUCT_EPOCH_RSP (820, P0, rpcTimeout, FETCH_METADATA, MessageSerializers::logStateSerializer, () -> ResponseVerbHandler.instance ),
TCM_RECONSTRUCT_EPOCH_REQ (821, P0, rpcTimeout, FETCH_METADATA, () -> ReconstructLogState.serializer, () -> ReconstructLogState.Handler.instance, TCM_FETCH_PEER_LOG_RSP ),

3
src/java/org/apache/cassandra/service/accord/AccordService.java
View File

@ -399,9 +399,10 @@ public class AccordService implements IAccordService, Shutdownable
int waitSeconds = 5;
while (true)
{
Epoch await = Epoch.max(Epoch.create(configService.currentEpoch()), metadata.epoch);
try
{
epochReady(metadata.epoch).get(waitSeconds, SECONDS);
epochReady(await).get(waitSeconds, SECONDS);
break;
}
catch (TimeoutException e)

6
src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java
View File

@ -216,6 +216,9 @@ public abstract class AbstractReadExecutor
if (retry.equals(NeverSpeculativeRetryPolicy.INSTANCE) || consistencyLevel == ConsistencyLevel.EACH_QUORUM)
return new NeverSpeculatingReadExecutor(coordinator, cfs, command, replicaPlan, requestTime, false);
if (retry.equals(AlwaysSpeculativeRetryPolicy.INSTANCE))
return new AlwaysSpeculatingReadExecutor(coordinator, cfs, command, replicaPlan, requestTime);
// There are simply no extra replicas to speculate.
// Handle this separately so it can record failed attempts to speculate due to lack of replicas
if (replicaPlan.contacts().size() == replicaPlan.readCandidates().size())
@ -223,9 +226,6 @@ public abstract class AbstractReadExecutor
boolean recordFailedSpeculation = consistencyLevel != ConsistencyLevel.ALL;
return new NeverSpeculatingReadExecutor(coordinator, cfs, command, replicaPlan, requestTime, recordFailedSpeculation);
}
if (retry.equals(AlwaysSpeculativeRetryPolicy.INSTANCE))
return new AlwaysSpeculatingReadExecutor(coordinator, cfs, command, replicaPlan, requestTime);
else // PERCENTILE or CUSTOM.
return new SpeculatingReadExecutor(coordinator, cfs, command, replicaPlan, requestTime);
}

33
test/distributed/org/apache/cassandra/distributed/test/log/FetchLogFromPeers2Test.java
View File

@ -19,7 +19,6 @@
package org.apache.cassandra.distributed.test.log;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import org.junit.Test;
@ -31,35 +30,39 @@ import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.Epoch;
import static org.apache.cassandra.distributed.test.log.FetchLogFromPeersTest.*;
import static org.apache.cassandra.distributed.test.log.FetchLogFromPeersTest.ClusterState;
import static org.apache.cassandra.distributed.test.log.FetchLogFromPeersTest.Operation;
import static org.apache.cassandra.distributed.test.log.FetchLogFromPeersTest.coordinator;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class FetchLogFromPeers2Test extends TestBaseImpl
{
@Test
public void testSchema() throws Exception
public void testSchema() throws Throwable
{
try (Cluster cluster = init(builder().withNodes(3)
.start()))
try (Cluster cluster = init(builder().withNodes(3).start()))
{
cluster.schemaChange(withKeyspace("alter keyspace %s with replication = {'class':'SimpleStrategy', 'replication_factor':3}"));
cluster.schemaChange(withKeyspace("create table %s.tbl (id int primary key)"));
cluster.schemaChange(withKeyspace("create table %s.tbl2 (id int primary key)"));
cluster.schemaChange(withKeyspace("alter keyspace %s with replication = {'class':'SimpleStrategy', 'replication_factor':3} "));
cluster.schemaChange(withKeyspace("create table %s.tbl (id int primary key) WITH speculative_retry = 'ALWAYS';"));
for (ClusterState clusterState : ClusterState.values())
{
for (Operation operation : Operation.values())
{
cluster.filters().inbound().from(1, 2).to(1, 2).drop();
setupSchemaBehind(cluster);
cluster.filters().inbound().to(1).to(1).drop();
runQuery(cluster, clusterState, operation);
cluster.filters().reset();
}
}
}
}
public void runQuery(Cluster cluster, ClusterState clusterState, Operation operation) throws ExecutionException, InterruptedException
public void runQuery(Cluster cluster, ClusterState clusterState, Operation operation) throws Throwable
{
cluster.get(1).shutdown().get();
// node2 is behind
String query;
switch (operation)
@ -78,7 +81,7 @@ public class FetchLogFromPeers2Test extends TestBaseImpl
long metricsBefore = cluster.get(2).callOnInstance(() -> TCMMetrics.instance.fetchedPeerLogEntries.getCount());
if (clusterState == ClusterState.COORDINATOR_BEHIND)
{
long [] coordinatorBehindMetricsBefore = new long[cluster.size()];
long[] coordinatorBehindMetricsBefore = new long[cluster.size()];
try
{
for (int i = 1; i <= cluster.size(); i++)
@ -102,20 +105,15 @@ public class FetchLogFromPeers2Test extends TestBaseImpl
}
}
assertTrue("Metric CoordinatorBehindSchema should have been bumped for at least one replica", metricBumped);
}
cluster.coordinator(coordinator).execute(withKeyspace(query), ConsistencyLevel.QUORUM);
assertTrue(cluster.get(2).logs().grep(mark, "Fetching log from /127.0.0.3:7012").getResult().size() > 0);
long metricsAfter = cluster.get(2).callOnInstance(() -> TCMMetrics.instance.fetchedPeerLogEntries.getCount());
assertTrue(metricsAfter > metricsBefore);
cluster.get(1).startup();
}
public void setupSchemaBehind(Cluster cluster)
{
cluster.filters().reset();
cluster.filters().inbound().from(1).to(2).drop();
long epochBefore = cluster.get(3).callOnInstance(() -> ClusterMetadata.current().epoch.getEpoch());
cluster.coordinator(1).execute(withKeyspace("alter table %s.tbl with comment='test " + UUID.randomUUID() + "'"), ConsistencyLevel.ONE);
cluster.get(3).runOnInstance(() -> {
@ -128,6 +126,5 @@ public class FetchLogFromPeers2Test extends TestBaseImpl
throw new RuntimeException(e);
}
});
cluster.filters().reset();
}
}