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

Merge branch 'cassandra-4.0' into trunk

This commit is contained in:
Marcus Eriksson 2022-01-17 14:13:41 +01:00
parent df1a2d4c3d 9a1bb62822
commit 96c80f0b89
6 changed files with 239 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -106,6 +106,7 @@ Merged from 3.11:
* Update Jackson from 2.9.10 to 2.12.5 (CASSANDRA-16851)
* Make assassinate more resilient to missing tokens (CASSANDRA-16847)
Merged from 3.0:
* Avoid race in AbstractReplicationStrategy endpoint caching (CASSANDRA-16673)
* Fix abort when window resizing during cqlsh COPY (CASSANDRA-15230)
* Fix slow keycache load which blocks startup for tables with many sstables (CASSANDRA-14898)
* Fix rare NPE caused by batchlog replay / node decomission races (CASSANDRA-17049)

92
src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
View File

@ -23,6 +23,7 @@ import java.lang.reflect.Method;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
import com.google.common.base.Preconditions;
@ -54,10 +55,7 @@ public abstract class AbstractReplicationStrategy
public final Map<String, String> configOptions;
protected final String keyspaceName;
private final TokenMetadata tokenMetadata;
// track when the token range changes, signaling we need to invalidate our endpoint cache
private volatile long lastInvalidatedVersion = 0;
private final ReplicaCache<Token, EndpointsForRange> replicas = new ReplicaCache<>();
public IEndpointSnitch snitch;
protected AbstractReplicationStrategy(String keyspaceName, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map<String, String> configOptions)
@ -70,26 +68,9 @@ public abstract class AbstractReplicationStrategy
this.keyspaceName = keyspaceName;
}
private final Map<Token, EndpointsForRange> cachedReplicas = new NonBlockingHashMap<>();
public EndpointsForRange getCachedReplicas(Token t)
public EndpointsForRange getCachedReplicas(long ringVersion, Token t)
{
long lastVersion = tokenMetadata.getRingVersion();
if (lastVersion > lastInvalidatedVersion)
{
synchronized (this)
{
if (lastVersion > lastInvalidatedVersion)
{
logger.trace("clearing cached endpoints");
cachedReplicas.clear();
lastInvalidatedVersion = lastVersion;
}
}
}
return cachedReplicas.get(t);
return replicas.get(ringVersion, t);
}
/**
@ -107,15 +88,16 @@ public abstract class AbstractReplicationStrategy
public EndpointsForRange getNaturalReplicas(RingPosition<?> searchPosition)
{
Token searchToken = searchPosition.getToken();
long currentRingVersion = tokenMetadata.getRingVersion();
Token keyToken = TokenMetadata.firstToken(tokenMetadata.sortedTokens(), searchToken);
EndpointsForRange endpoints = getCachedReplicas(keyToken);
EndpointsForRange endpoints = getCachedReplicas(currentRingVersion, keyToken);
if (endpoints == null)
{
TokenMetadata tm = tokenMetadata.cachedOnlyTokenMap();
// if our cache got invalidated, it's possible there is a new token to account for too
keyToken = TokenMetadata.firstToken(tm.sortedTokens(), searchToken);
endpoints = calculateNaturalReplicas(searchToken, tm);
cachedReplicas.put(keyToken, endpoints);
replicas.put(tm.getRingVersion(), keyToken, endpoints);
}
return endpoints;
@ -463,4 +445,64 @@ public abstract class AbstractReplicationStrategy
throw new ConfigurationException(String.format("Unrecognized strategy option {%s} passed to %s for keyspace %s", key, getClass().getSimpleName(), keyspaceName));
}
}
static class ReplicaCache<K, V>
{
private final AtomicReference<ReplicaHolder<K, V>> cachedReplicas = new AtomicReference<>(new ReplicaHolder<>(0, 4));
V get(long ringVersion, K keyToken)
{
ReplicaHolder<K, V> replicaHolder = maybeClearAndGet(ringVersion);
if (replicaHolder == null)
return null;
return replicaHolder.replicas.get(keyToken);
}
void put(long ringVersion, K keyToken, V endpoints)
{
ReplicaHolder<K, V> current = maybeClearAndGet(ringVersion);
if (current != null)
{
// if we have the same ringVersion, but already know about the keyToken the endpoints should be the same
current.replicas.putIfAbsent(keyToken, endpoints);
}
}
ReplicaHolder<K, V> maybeClearAndGet(long ringVersion)
{
ReplicaHolder<K, V> current = cachedReplicas.get();
if (ringVersion == current.ringVersion)
return current;
else if (ringVersion < current.ringVersion) // things have already moved on
return null;
// If ring version has changed, create a fresh replica holder and try to replace the current one.
// This may race with other threads that have the same new ring version and one will win and the loosers
// will be garbage collected
ReplicaHolder<K, V> cleaned = new ReplicaHolder<>(ringVersion, current.replicas.size());
cachedReplicas.compareAndSet(current, cleaned);
// A new ring version may have come along while making the new holder, so re-check the
// reference and return the ring version if the same, otherwise return null as there is no point
// in using it.
current = cachedReplicas.get();
if (ringVersion == current.ringVersion)
return current;
else
return null;
}
}
static class ReplicaHolder<K, V>
{
private final long ringVersion;
private final NonBlockingHashMap<K, V> replicas;
ReplicaHolder(long ringVersion, int expectedEntries)
{
this.ringVersion = ringVersion;
this.replicas = new NonBlockingHashMap<>(expectedEntries);
}
}
}

9
src/java/org/apache/cassandra/locator/TokenMetadata.java
View File

@ -131,12 +131,18 @@ public class TokenMetadata
}
private TokenMetadata(BiMultiValMap<Token, InetAddressAndPort> tokenToEndpointMap, BiMap<InetAddressAndPort, UUID> endpointsMap, Topology topology, IPartitioner partitioner)
{
this(tokenToEndpointMap, endpointsMap, topology, partitioner, 0);
}
private TokenMetadata(BiMultiValMap<Token, InetAddressAndPort> tokenToEndpointMap, BiMap<InetAddressAndPort, UUID> endpointsMap, Topology topology, IPartitioner partitioner, long ringVersion)
{
this.tokenToEndpointMap = tokenToEndpointMap;
this.topology = topology;
this.partitioner = partitioner;
endpointToHostIdMap = endpointsMap;
sortedTokens = sortTokens();
this.ringVersion = ringVersion;
}
/**
@ -675,7 +681,8 @@ public class TokenMetadata
return new TokenMetadata(SortedBiMultiValMap.create(tokenToEndpointMap),
HashBiMap.create(endpointToHostIdMap),
topology,
partitioner);
partitioner,
ringVersion);
}
finally
{

6
test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java
View File

@ -56,7 +56,7 @@ public class BootstrapTest extends TestBaseImpl
.withConfig(config -> config.with(NETWORK, GOSSIP))
.start())
{
populate(cluster,0, 100);
populate(cluster, 0, 100);
IInstanceConfig config = cluster.newInstanceConfig();
IInvokableInstance newInstance = cluster.bootstrap(config);
@ -95,7 +95,7 @@ public class BootstrapTest extends TestBaseImpl
cluster.forEach(statusToBootstrap(newInstance));
populate(cluster,0, 100);
populate(cluster, 0, 100);
Assert.assertEquals(100, newInstance.executeInternal("SELECT *FROM " + KEYSPACE + ".tbl").length);
}
@ -113,7 +113,7 @@ public class BootstrapTest extends TestBaseImpl
.withConfig(config -> config.with(NETWORK, GOSSIP))
.start())
{
populate(cluster,0, 100);
populate(cluster, 0, 100);
bootstrapAndJoinNode(cluster);
for (Map.Entry<Integer, Long> e : count(cluster).entrySet())

115
test/distributed/org/apache/cassandra/distributed/test/ring/ReadsDuringBootstrapTest.java Normal file
View File

@ -0,0 +1,115 @@
/*
* 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.ring;
import java.io.IOException;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
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.FieldValue;
import net.bytebuddy.implementation.bind.annotation.SuperCall;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.TokenSupplier;
import org.apache.cassandra.distributed.shared.NetworkTopology;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.locator.EndpointsForRange;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
import static org.apache.cassandra.distributed.api.Feature.NETWORK;
public class ReadsDuringBootstrapTest extends TestBaseImpl
{
@Test
public void readsDuringBootstrapTest() throws IOException, ExecutionException, InterruptedException, TimeoutException
{
int originalNodeCount = 3;
int expandedNodeCount = originalNodeCount + 1;
ExecutorService es = Executors.newSingleThreadExecutor();
try (Cluster cluster = builder().withNodes(originalNodeCount)
.withTokenSupplier(TokenSupplier.evenlyDistributedTokens(expandedNodeCount))
.withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(expandedNodeCount, "dc0", "rack0"))
.withConfig(config -> config.with(NETWORK, GOSSIP)
.set("read_request_timeout_in_ms", Integer.MAX_VALUE)
.set("request_timeout_in_ms", Integer.MAX_VALUE))
.withInstanceInitializer(BB::install)
.start())
{
String query = withKeyspace("SELECT * FROM %s.tbl WHERE id = ?");
cluster.schemaChange(withKeyspace("CREATE KEYSPACE IF NOT EXISTS %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 2};"));
cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (id int PRIMARY KEY)"));
cluster.get(1).runOnInstance(() -> BB.block.set(true));
Future<?> read = es.submit(() -> cluster.coordinator(1).execute(query, ConsistencyLevel.QUORUM, 3));
long mark = cluster.get(1).logs().mark();
bootstrapAndJoinNode(cluster);
cluster.get(1).logs().watchFor(mark, "New node /127.0.0.4");
cluster.get(1).runOnInstance(() -> BB.block.set(false));
// populate cache
for (int i = 0; i < 10; i++)
cluster.coordinator(1).execute(query, ConsistencyLevel.QUORUM, i);
cluster.get(1).runOnInstance(() -> BB.latch.countDown());
read.get();
}
finally
{
es.shutdown();
}
}
public static class BB
{
public static final AtomicBoolean block = new AtomicBoolean();
public static final CountDownLatch latch = new CountDownLatch(1);
private static void install(ClassLoader cl, Integer instanceId)
{
if (instanceId != 1)
return;
new ByteBuddy().rebase(AbstractReplicationStrategy.class)
.method(named("getCachedReplicas"))
.intercept(MethodDelegation.to(BB.class))
.make()
.load(cl, ClassLoadingStrategy.Default.INJECTION);
}
public static EndpointsForRange getCachedReplicas(long ringVersion, Token t,
@FieldValue("keyspaceName") String keyspaceName,
@SuperCall Callable<EndpointsForRange> zuper) throws Exception
{
if (keyspaceName.equals(KEYSPACE) && block.get())
latch.await();
return zuper.call();
}
}
}

45
test/unit/org/apache/cassandra/locator/AbstractReplicationStrategyTest.java Normal file
View File

@ -0,0 +1,45 @@
/*
* 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.locator;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
public class AbstractReplicationStrategyTest
{
@Test
public void testReplicaCache()
{
AbstractReplicationStrategy.ReplicaCache<Integer, Integer> cache = new AbstractReplicationStrategy.ReplicaCache<>();
cache.put(10, 1, 1);
assertEquals(1, (int)cache.get(10, 1));
assertNull(cache.get(9,1)); // get with old ringversion, return null to force a recalculation
assertNull(cache.get(11,1)); // newer ringVersion - cache gets cleared
assertNull(cache.get(10,1)); // and make sure the map got cleared
cache.put(11, 1, 100);
cache.put(10, 1, 99);
assertEquals(100, (int)cache.get(11, 1));
assertNull(cache.get(12, 55));
assertNull(cache.get(11, 1));
}
}