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

CASSANDRA-19633 Replaced node is stuck in a loop calculating ranges

This commit is contained in:
Bereng 2025-04-10 13:33:33 +02:00
parent 96bfbe6250
commit fcea0b6fd8
3 changed files with 84 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
View File

@ -259,8 +259,13 @@ public enum CassandraRelevantProperties
/**
* Number of replicas required to store batchlog for atomicity, only accepts values of 1 or 2.
*/
REQUIRED_BATCHLOG_REPLICA_COUNT("cassandra.batchlog.required_replica_count", "2")
;
REQUIRED_BATCHLOG_REPLICA_COUNT("cassandra.batchlog.required_replica_count", "2"),
/**
* Do not try to calculate optimal streaming candidates. This can take a lot of time in some configs specially
* with vnodes.
*/
SKIP_OPTIMAL_STREAMING_CANDIDATES_CALCULATION("cassandra.skip_optimal_streaming_candidates_calculation", "false");
CassandraRelevantProperties(String key, String defaultVal)
{

9
src/java/org/apache/cassandra/dht/RangeStreamer.java
View File

@ -41,6 +41,7 @@ import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.gms.FailureDetector;
@ -331,8 +332,12 @@ public class RangeStreamer
Multimap<InetAddressAndPort, FetchReplica> workMap;
//Only use the optimized strategy if we don't care about strict sources, have a replication factor > 1, and no
//transient replicas.
if (useStrictSource || strat == null || strat.getReplicationFactor().allReplicas == 1 || strat.getReplicationFactor().hasTransientReplicas())
//transient replicas or it is intentionally skipped.
if (CassandraRelevantProperties.SKIP_OPTIMAL_STREAMING_CANDIDATES_CALCULATION.getBoolean() ||
useStrictSource ||
strat == null ||
strat.getReplicationFactor().allReplicas == 1 ||
strat.getReplicationFactor().hasTransientReplicas())
{
workMap = convertPreferredEndpointsToWorkMap(fetchMap);
}

86
test/unit/org/apache/cassandra/dht/BootStrapperTest.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.dht;
import java.net.UnknownHostException;
import java.util.List;
import java.util.Random;
import java.util.concurrent.atomic.AtomicBoolean;
import com.google.common.base.Predicate;
import com.google.common.base.Predicates;
@ -28,8 +29,10 @@ import com.google.common.collect.Multimap;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.RangeStreamer.FetchReplica;
@ -42,16 +45,35 @@ import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.streaming.StreamOperation;
import org.jboss.byteman.contrib.bmunit.BMRule;
import org.jboss.byteman.contrib.bmunit.BMRules;
import org.jboss.byteman.contrib.bmunit.BMUnitRunner;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
@RunWith(BMUnitRunner.class)
public class BootStrapperTest
{
static IPartitioner oldPartitioner;
static Predicate<Replica> originalAlivePredicate = RangeStreamer.ALIVE_PREDICATE;
public static AtomicBoolean nonOptimizationHit = new AtomicBoolean(false);
public static AtomicBoolean optimizationHit = new AtomicBoolean(false);
private static final IFailureDetector mockFailureDetector = new IFailureDetector()
{
public boolean isAlive(InetAddressAndPort ep)
{
return true;
}
public void interpret(InetAddressAndPort ep) { throw new UnsupportedOperationException(); }
public void report(InetAddressAndPort ep) { throw new UnsupportedOperationException(); }
public void registerFailureDetectionEventListener(IFailureDetectionEventListener listener) { throw new UnsupportedOperationException(); }
public void unregisterFailureDetectionEventListener(IFailureDetectionEventListener listener) { throw new UnsupportedOperationException(); }
public void remove(InetAddressAndPort ep) { throw new UnsupportedOperationException(); }
public void forceConviction(InetAddressAndPort ep) { throw new UnsupportedOperationException(); }
};
@BeforeClass
public static void setup() throws ConfigurationException
{
@ -83,6 +105,52 @@ public class BootStrapperTest
}
}
@Test
@BMRules(rules = { @BMRule(name = "Make sure the non-optimized path is picked up for some operations",
targetClass = "org.apache.cassandra.dht.RangeStreamer",
targetMethod = "convertPreferredEndpointsToWorkMap(EndpointsByReplica)",
action = "org.apache.cassandra.dht.BootStrapperTest.nonOptimizationHit.set(true)"),
@BMRule(name = "Make sure the optimized path is picked up for some operations",
targetClass = "org.apache.cassandra.dht.RangeStreamer",
targetMethod = "getOptimizedWorkMap(EndpointsByReplica,Collection,String)",
action = "org.apache.cassandra.dht.BootStrapperTest.optimizationHit.set(true)") })
public void testStreamingCandidatesOptmizationSkip() throws UnknownHostException
{
testSkipStreamingCandidatesOptmizationFeatureFlag(true, true, false);
testSkipStreamingCandidatesOptmizationFeatureFlag(false, true, true);
}
private void testSkipStreamingCandidatesOptmizationFeatureFlag(boolean disableOptimization, boolean nonOptimizedPathHit, boolean optimizedPathHit) throws UnknownHostException
{
try
{
nonOptimizationHit.set(false);
optimizationHit.set(false);
CassandraRelevantProperties.SKIP_OPTIMAL_STREAMING_CANDIDATES_CALCULATION.setBoolean(disableOptimization);
for (String keyspaceName : Schema.instance.getNonLocalStrategyKeyspaces())
{
StorageService ss = StorageService.instance;
TokenMetadata tmd = ss.getTokenMetadata();
generateFakeEndpoints(10);
Token myToken = tmd.partitioner.getRandomToken();
InetAddressAndPort myEndpoint = InetAddressAndPort.getByName("127.0.0.1");
assertEquals(10, tmd.sortedTokens().size());
RangeStreamer s = new RangeStreamer(tmd, null, myEndpoint, StreamOperation.BOOTSTRAP, true, DatabaseDescriptor.getEndpointSnitch(), new StreamStateStore(), mockFailureDetector, false, 1);
s.addRanges(keyspaceName, Keyspace.open(keyspaceName).getReplicationStrategy().getPendingAddressRanges(tmd, myToken, myEndpoint));
}
assertEquals(nonOptimizedPathHit, nonOptimizationHit.get());
assertEquals(optimizedPathHit, optimizationHit.get());
}
finally
{
CassandraRelevantProperties.SKIP_OPTIMAL_STREAMING_CANDIDATES_CALCULATION.reset();
}
}
private RangeStreamer testSourceTargetComputation(String keyspaceName, int numOldNodes, int replicationFactor) throws UnknownHostException
{
StorageService ss = StorageService.instance;
@ -93,20 +161,6 @@ public class BootStrapperTest
InetAddressAndPort myEndpoint = InetAddressAndPort.getByName("127.0.0.1");
assertEquals(numOldNodes, tmd.sortedTokens().size());
IFailureDetector mockFailureDetector = new IFailureDetector()
{
public boolean isAlive(InetAddressAndPort ep)
{
return true;
}
public void interpret(InetAddressAndPort ep) { throw new UnsupportedOperationException(); }
public void report(InetAddressAndPort ep) { throw new UnsupportedOperationException(); }
public void registerFailureDetectionEventListener(IFailureDetectionEventListener listener) { throw new UnsupportedOperationException(); }
public void unregisterFailureDetectionEventListener(IFailureDetectionEventListener listener) { throw new UnsupportedOperationException(); }
public void remove(InetAddressAndPort ep) { throw new UnsupportedOperationException(); }
public void forceConviction(InetAddressAndPort ep) { throw new UnsupportedOperationException(); }
};
RangeStreamer s = new RangeStreamer(tmd, null, myEndpoint, StreamOperation.BOOTSTRAP, true, DatabaseDescriptor.getEndpointSnitch(), new StreamStateStore(), mockFailureDetector, false, 1);
assertNotNull(Keyspace.open(keyspaceName));
s.addRanges(keyspaceName, Keyspace.open(keyspaceName).getReplicationStrategy().getPendingAddressRanges(tmd, myToken, myEndpoint));