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

Accord: Retry epoch/topology metadata fetch on all peer nodes 

Patch by Alex Petrov; reviewed by David Capwell for CASSANDRA-20663
This commit is contained in:
Alex Petrov 2025-05-20 12:12:01 +02:00
parent a739e795de
commit 66b973341a
3 changed files with 33 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
src/java/org/apache/cassandra/gms/FailureDetector.java
View File

@ -326,7 +326,7 @@ public class FailureDetector implements IFailureDetector, FailureDetectorMBean
// an error in that case.
ClusterMetadata metadata = ClusterMetadata.current();
if (!metadata.directory.allJoinedEndpoints().contains(ep) && !metadata.fullCMSMembers().contains(ep))
logger.error("Unknown endpoint: " + ep, new IllegalArgumentException("Unknown endpoint: " + ep));
logger.error("Unknown endpoint: " + ep, new UnknownEndpointException(ep));
}
return epState != null && epState.isAlive();
}
@ -437,6 +437,14 @@ public class FailureDetector implements IFailureDetector, FailureDetectorMBean
sb.append("-----------------------------------------------------------------------");
return sb.toString();
}
public static class UnknownEndpointException extends IllegalArgumentException
{
public UnknownEndpointException(InetAddressAndPort ep)
{
super("Unknown endpoint: " + ep);
}
}
}
/*

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

@ -448,35 +448,30 @@ public class AccordService implements IAccordService, Shutdownable
if (peers.isEmpty())
return null;
Iterator<InetAddressAndPort> iter = peers.iterator();
while (iter.hasNext())
try
{
InetAddressAndPort peer = iter.next();
try
{
logger.info("Fetching topologies for epochs [{}, {}] from {}", from, metadata.epoch.getEpoch(), peer);
Invariants.require(from <= metadata.epoch.getEpoch(),
"Accord epochs should never be ahead of TCM ones, but %d was ahead of %d", from, metadata.epoch.getEpoch());
logger.info("Fetching topologies for epochs [{}, {}] from {}", from, metadata.epoch.getEpoch(), peers);
Invariants.require(from <= metadata.epoch.getEpoch(),
"Accord epochs should never be ahead of TCM ones, but %d was ahead of %d", from, metadata.epoch.getEpoch());
Future<TopologyRange> futures = FetchTopologies.fetch(SharedContext.Global.instance,
Collections.singleton(peer),
from,
Long.MAX_VALUE);
TopologyRange response = futures.get();
logger.info("Fetched topologies {}", response);
Future<TopologyRange> futures = FetchTopologies.fetch(SharedContext.Global.instance,
peers,
from,
Long.MAX_VALUE);
TopologyRange response = futures.get();
logger.info("Fetched topologies {}", response);
// We're behind and need to catch up CMS first.
if (response.current > ClusterMetadata.current().epoch.getEpoch())
ClusterMetadataService.instance().fetchLogFromCMS(Epoch.create(response.current));
// We're behind and need to catch up CMS first.
if (response.current > ClusterMetadata.current().epoch.getEpoch())
ClusterMetadataService.instance().fetchLogFromCMS(Epoch.create(response.current));
if (response.current >= from)
return response;
metadata = ClusterMetadata.current();
}
catch (Throwable e)
{
logger.info("Failed to fetch epochs [{}, {}] from {}", from, metadata.epoch.getEpoch(), peer);
}
if (response.current >= from)
return response;
metadata = ClusterMetadata.current();
}
catch (Throwable e)
{
logger.info("Failed to fetch epochs [{}, {}] from {}", from, metadata.epoch.getEpoch(), peers);
}
// After trying to contact all peers, and retrying according to retry spec on them, we give up.

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

@ -35,6 +35,7 @@ import java.util.stream.Stream;
import javax.annotation.Nullable;
import com.google.common.base.Throwables;
import org.apache.cassandra.gms.FailureDetector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -85,7 +86,8 @@ public abstract class PaxosSimulation implements Simulation, ClusterActionListen
protected Class<? extends Throwable>[] expectedExceptionsPaxos()
{
return (Class<? extends Throwable>[]) new Class<?>[] { RequestExecutionException.class,
CancellationException.class };
CancellationException.class,
FailureDetector.UnknownEndpointException.class};
}
@SuppressWarnings("unchecked")
@ -97,6 +99,7 @@ public abstract class PaxosSimulation implements Simulation, ClusterActionListen
CancellationException.class,
CoordinationFailed.class,
ClosedChannelException.class,
FailureDetector.UnknownEndpointException.class,
StreamReceivedOutOfTokenRangeException.class // should always come in combination with closed channel exception
};
}