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

Merge branch 'cassandra-3.0' into cassandra-3.11

This commit is contained in:
Joel Knighton 2017-03-22 13:20:24 -05:00
parent 5484bd1acc 2836a644a3
commit ec9ce3dfba
8 changed files with 252 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -38,6 +38,7 @@ Merged from 3.0:
* Provide user workaround when system_schema.columns does not contain entries
for a table that's in system_schema.tables (CASSANDRA-13180)
Merged from 2.2:
* Discard in-flight shadow round responses (CASSANDRA-12653)
* Don't anti-compact repaired data to avoid inconsistencies (CASSANDRA-13153)
* Wrong logger name in AnticompactionTask (CASSANDRA-13343)
* Commitlog replay may fail if last mutation is within 4 bytes of end of segment (CASSANDRA-13282)

27
src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java
View File

@ -51,22 +51,33 @@ public class GossipDigestAckVerbHandler implements IVerbHandler<GossipDigestAck>
Map<InetAddress, EndpointState> epStateMap = gDigestAckMessage.getEndpointStateMap();
logger.trace("Received ack with {} digests and {} states", gDigestList.size(), epStateMap.size());
if (epStateMap.size() > 0)
{
/* Notify the Failure Detector */
Gossiper.instance.notifyFailureDetector(epStateMap);
Gossiper.instance.applyStateLocally(epStateMap);
}
if (Gossiper.instance.isInShadowRound())
{
if (logger.isDebugEnabled())
logger.debug("Received an ack from {}, which may trigger exit from shadow round", from);
// if the ack is completely empty, then we can infer that the respondent is also in a shadow round
Gossiper.instance.maybeFinishShadowRound(from, gDigestList.isEmpty() && epStateMap.isEmpty());
Gossiper.instance.maybeFinishShadowRound(from, gDigestList.isEmpty() && epStateMap.isEmpty(), epStateMap);
return; // don't bother doing anything else, we have what we came for
}
if (epStateMap.size() > 0)
{
// Ignore any GossipDigestAck messages that we handle before a regular GossipDigestSyn has been send.
// This will prevent Acks from leaking over from the shadow round that are not actual part of
// the regular gossip conversation.
if ((System.nanoTime() - Gossiper.instance.firstSynSendAt) < 0 || Gossiper.instance.firstSynSendAt == 0)
{
if (logger.isTraceEnabled())
logger.trace("Ignoring unrequested GossipDigestAck from {}", from);
return;
}
/* Notify the Failure Detector */
Gossiper.instance.notifyFailureDetector(epStateMap);
Gossiper.instance.applyStateLocally(epStateMap);
}
/* Get the state required to send to this gossipee - construct GossipDigestAck2Message */
Map<InetAddress, EndpointState> deltaEpStateMap = new HashMap<InetAddress, EndpointState>();
for (GossipDigest gDigest : gDigestList)

65
src/java/org/apache/cassandra/gms/Gossiper.java
View File

@ -30,6 +30,7 @@ import javax.management.ObjectName;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.cassandra.utils.CassandraVersion;
@ -88,6 +89,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
private static final Logger logger = LoggerFactory.getLogger(Gossiper.class);
public static final Gossiper instance = new Gossiper();
// Timestamp to prevent processing any in-flight messages for we've not send any SYN yet, see CASSANDRA-12653.
volatile long firstSynSendAt = 0L;
public static final long aVeryLongTime = 259200 * 1000; // 3 days
// Maximimum difference between generation value and local time we are willing to accept about a peer
@ -112,7 +116,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
private final Map<InetAddress, Long> unreachableEndpoints = new ConcurrentHashMap<InetAddress, Long>();
/* initial seeds for joining the cluster */
private final Set<InetAddress> seeds = new ConcurrentSkipListSet<InetAddress>(inetcomparator);
@VisibleForTesting
final Set<InetAddress> seeds = new ConcurrentSkipListSet<InetAddress>(inetcomparator);
/* map where key is the endpoint and value is the state associated with the endpoint */
final ConcurrentMap<InetAddress, EndpointState> endpointStateMap = new ConcurrentHashMap<InetAddress, EndpointState>();
@ -126,7 +131,10 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
private final Map<InetAddress, Long> expireTimeEndpointMap = new ConcurrentHashMap<InetAddress, Long>();
private volatile boolean inShadowRound = false;
// seeds gathered during shadow round that indicated to be in the shadow round phase as well
private final Set<InetAddress> seedsInShadowRound = new ConcurrentSkipListSet<>(inetcomparator);
// endpoint states as gathered during shadow round
private final Map<InetAddress, EndpointState> endpointShadowStateMap = new ConcurrentHashMap<>();
private volatile long lastProcessedMessageAt = System.currentTimeMillis();
@ -650,6 +658,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
InetAddress to = liveEndpoints.get(index);
if (logger.isTraceEnabled())
logger.trace("Sending a GossipDigestSyn to {} ...", to);
if (firstSynSendAt == 0)
firstSynSendAt = System.nanoTime();
MessagingService.instance().sendOneWay(message, to);
return seeds.contains(to);
}
@ -726,11 +736,13 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
* @param endpoint - the endpoint to check
* @param localHostUUID - the host id to check
* @param isBootstrapping - whether the node intends to bootstrap when joining
* @param epStates - endpoint states in the cluster
* @return true if it is safe to start the node, false otherwise
*/
public boolean isSafeForStartup(InetAddress endpoint, UUID localHostUUID, boolean isBootstrapping)
public boolean isSafeForStartup(InetAddress endpoint, UUID localHostUUID, boolean isBootstrapping,
Map<InetAddress, EndpointState> epStates)
{
EndpointState epState = endpointStateMap.get(endpoint);
EndpointState epState = epStates.get(endpoint);
// if there's no previous state, or the node was previously removed from the cluster, we're good
if (epState == null || isDeadState(epState))
return true;
@ -854,14 +866,6 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
return !(value1 == null || value2 == null) && value1.value.equals(value2.value);
}
// removes ALL endpoint states; should only be called after shadow gossip
public void resetEndpointStateMap()
{
endpointStateMap.clear();
unreachableEndpoints.clear();
liveEndpoints.clear();
}
public Set<Entry<InetAddress, EndpointState>> getEndpointStates()
{
return endpointStateMap.entrySet();
@ -869,7 +873,12 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
public UUID getHostId(InetAddress endpoint)
{
return UUID.fromString(getEndpointStateForEndpoint(endpoint).getApplicationState(ApplicationState.HOST_ID).value);
return getHostId(endpoint, endpointStateMap);
}
public UUID getHostId(InetAddress endpoint, Map<InetAddress, EndpointState> epStates)
{
return UUID.fromString(epStates.get(endpoint).getApplicationState(ApplicationState.HOST_ID).value);
}
EndpointState getStateForVersionBiggerThan(InetAddress forEndpoint, int version)
@ -1343,20 +1352,32 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
}
/**
* Do a single 'shadow' round of gossip, where we do not modify any state
* Used when preparing to join the ring:
* * when replacing a node, to get and assume its tokens
* * when joining, to check that the local host id matches any previous id for the endpoint address
* Do a single 'shadow' round of gossip by retrieving endpoint states that will be stored exclusively in the
* map return value, instead of endpointStateMap.
*
* <ul>
* <li>when replacing a node, to get and assume its tokens</li>
* <li>when joining, to check that the local host id matches any previous id for the endpoint address</li>
* </ul>
*
* Method is synchronized, as we use an in-progress flag to indicate that shadow round must be cleared
* again by calling {@link Gossiper#maybeFinishShadowRound(InetAddress, boolean, Map)}. This will update
* {@link Gossiper#endpointShadowStateMap} with received values, in order to return an immutable copy to the
* caller of {@link Gossiper#doShadowRound()}. Therefor only a single shadow round execution is permitted at
* the same time.
*
* @return endpoint states gathered during shadow round or empty map
*/
public void doShadowRound()
public synchronized Map<InetAddress, EndpointState> doShadowRound()
{
buildSeedsList();
// it may be that the local address is the only entry in the seed
// list in which case, attempting a shadow round is pointless
if (seeds.isEmpty())
return;
return endpointShadowStateMap;
seedsInShadowRound.clear();
endpointShadowStateMap.clear();
// send a completely empty syn
List<GossipDigest> gDigests = new ArrayList<GossipDigest>();
GossipDigestSyn digestSynMessage = new GossipDigestSyn(DatabaseDescriptor.getClusterName(),
@ -1401,9 +1422,12 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
{
throw new RuntimeException(wtf);
}
return ImmutableMap.copyOf(endpointShadowStateMap);
}
private void buildSeedsList()
@VisibleForTesting
void buildSeedsList()
{
for (InetAddress seed : DatabaseDescriptor.getSeeds())
{
@ -1521,7 +1545,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
return (scheduledGossipTask != null) && (!scheduledGossipTask.isCancelled());
}
protected void maybeFinishShadowRound(InetAddress respondent, boolean isInShadowRound)
protected void maybeFinishShadowRound(InetAddress respondent, boolean isInShadowRound, Map<InetAddress, EndpointState> epStateMap)
{
if (inShadowRound)
{
@ -1529,6 +1553,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
{
logger.debug("Received a regular ack from {}, can now exit shadow round", respondent);
// respondent sent back a full ack, so we can exit our shadow round
endpointShadowStateMap.putAll(epStateMap);
inShadowRound = false;
seedsInShadowRound.clear();
}

12
src/java/org/apache/cassandra/service/MigrationTask.java
View File

@ -62,6 +62,12 @@ class MigrationTask extends WrappedRunnable
public void runMayThrow() throws Exception
{
if (!FailureDetector.instance.isAlive(endpoint))
{
logger.warn("Can't send schema pull request: node {} is down.", endpoint);
return;
}
// There is a chance that quite some time could have passed between now and the MM#maybeScheduleSchemaPull(),
// potentially enough for the endpoint node to restart - which is an issue if it does restart upgraded, with
// a higher major.
@ -71,12 +77,6 @@ class MigrationTask extends WrappedRunnable
return;
}
if (!FailureDetector.instance.isAlive(endpoint))
{
logger.debug("Can't send schema pull request: node {} is down.", endpoint);
return;
}
MessageOut message = new MessageOut<>(MessagingService.Verb.MIGRATION_REQUEST, null, MigrationManager.MigrationsSerializer.instance);
final CountDownLatch completionLatch = new CountDownLatch(1);

17
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -508,14 +508,14 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
InetAddress replaceAddress = DatabaseDescriptor.getReplaceAddress();
logger.info("Gathering node replacement information for {}", replaceAddress);
Gossiper.instance.doShadowRound();
Map<InetAddress, EndpointState> epStates = Gossiper.instance.doShadowRound();
// as we've completed the shadow round of gossip, we should be able to find the node we're replacing
if (Gossiper.instance.getEndpointStateForEndpoint(replaceAddress) == null)
if (epStates.get(replaceAddress) == null)
throw new RuntimeException(String.format("Cannot replace_address %s because it doesn't exist in gossip", replaceAddress));
try
{
VersionedValue tokensVersionedValue = Gossiper.instance.getEndpointStateForEndpoint(replaceAddress).getApplicationState(ApplicationState.TOKENS);
VersionedValue tokensVersionedValue = epStates.get(replaceAddress).getApplicationState(ApplicationState.TOKENS);
if (tokensVersionedValue == null)
throw new RuntimeException(String.format("Could not find tokens for %s to replace", replaceAddress));
@ -530,11 +530,10 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
if (isReplacingSameAddress())
{
localHostId = Gossiper.instance.getHostId(replaceAddress);
localHostId = Gossiper.instance.getHostId(replaceAddress, epStates);
SystemKeyspace.setLocalHostId(localHostId); // use the replacee's host Id as our own so we receive hints, etc
}
Gossiper.instance.resetEndpointStateMap(); // clean up since we have what we need
return localHostId;
}
@ -547,12 +546,12 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
}
logger.debug("Starting shadow gossip round to check for endpoint collision");
Gossiper.instance.doShadowRound();
Map<InetAddress, EndpointState> epStates = Gossiper.instance.doShadowRound();
// If bootstrapping, check whether any previously known status for the endpoint makes it unsafe to do so.
// If not bootstrapping, compare the host id for this endpoint learned from gossip (if any) with the local
// one, which was either read from system.local or generated at startup. If a learned id is present &
// doesn't match the local, then the node needs replacing
if (!Gossiper.instance.isSafeForStartup(FBUtilities.getBroadcastAddress(), localHostId, shouldBootstrap()))
if (!Gossiper.instance.isSafeForStartup(FBUtilities.getBroadcastAddress(), localHostId, shouldBootstrap(), epStates))
{
throw new RuntimeException(String.format("A node with address %s already exists, cancelling join. " +
"Use cassandra.replace_address if you want to replace this node.",
@ -561,7 +560,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
if (shouldBootstrap() && useStrictConsistency && !allowSimultaneousMoves())
{
for (Map.Entry<InetAddress, EndpointState> entry : Gossiper.instance.getEndpointStates())
for (Map.Entry<InetAddress, EndpointState> entry : epStates.entrySet())
{
// ignore local node or empty status
if (entry.getKey().equals(FBUtilities.getBroadcastAddress()) || entry.getValue().getApplicationState(ApplicationState.STATUS) == null)
@ -573,8 +572,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
throw new UnsupportedOperationException("Other bootstrapping/leaving/moving nodes detected, cannot bootstrap while cassandra.consistent.rangemovement is true");
}
}
logger.debug("Resetting gossip state after shadow round");
Gossiper.instance.resetEndpointStateMap();
}
private boolean allowSimultaneousMoves()

43
test/conf/cassandra-seeds.yaml Normal file
View File

@ -0,0 +1,43 @@
#
# Warning!
# Consider the effects on 'o.a.c.i.s.LegacySSTableTest' before changing schemas in this file.
#
cluster_name: Test Cluster
# memtable_allocation_type: heap_buffers
memtable_allocation_type: offheap_objects
commitlog_sync: batch
commitlog_sync_batch_window_in_ms: 1.0
commitlog_segment_size_in_mb: 5
commitlog_directory: build/test/cassandra/commitlog
cdc_raw_directory: build/test/cassandra/cdc_raw
cdc_enabled: false
hints_directory: build/test/cassandra/hints
partitioner: org.apache.cassandra.dht.ByteOrderedPartitioner
listen_address: 127.0.0.1
storage_port: 7010
start_native_transport: true
native_transport_port: 9042
column_index_size_in_kb: 4
saved_caches_directory: build/test/cassandra/saved_caches
data_file_directories:
- build/test/cassandra/data
disk_access_mode: mmap
seed_provider:
- class_name: org.apache.cassandra.locator.SimpleSeedProvider
parameters:
- seeds: "127.0.0.10,127.0.1.10,127.0.2.10"
endpoint_snitch: org.apache.cassandra.locator.SimpleSnitch
dynamic_snitch: true
server_encryption_options:
internode_encryption: none
keystore: conf/.keystore
keystore_password: cassandra
truststore: conf/.truststore
truststore_password: cassandra
incremental_backups: true
concurrent_compactors: 4
compaction_throughput_mb_per_sec: 0
row_cache_class_name: org.apache.cassandra.cache.OHCProvider
row_cache_size_in_mb: 16
enable_user_defined_functions: true
enable_scripted_user_defined_functions: true

116
test/unit/org/apache/cassandra/gms/ShadowRoundTest.java Normal file
View File

@ -0,0 +1,116 @@
/*
* 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.gms;
import java.util.Collections;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.After;
import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.locator.IEndpointSnitch;
import org.apache.cassandra.locator.PropertyFileSnitch;
import org.apache.cassandra.net.MessageIn;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.MockMessagingService;
import org.apache.cassandra.net.MockMessagingSpy;
import org.apache.cassandra.service.StorageService;
import static org.apache.cassandra.net.MockMessagingService.verb;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
public class ShadowRoundTest
{
private static final Logger logger = LoggerFactory.getLogger(ShadowRoundTest.class);
@BeforeClass
public static void setUp() throws ConfigurationException
{
System.setProperty("cassandra.config", "cassandra-seeds.yaml");
DatabaseDescriptor.daemonInitialization();
IEndpointSnitch snitch = new PropertyFileSnitch();
DatabaseDescriptor.setEndpointSnitch(snitch);
Keyspace.setInitialized();
}
@After
public void cleanup()
{
MockMessagingService.cleanup();
}
@Test
public void testDelayedResponse()
{
Gossiper.instance.buildSeedsList();
int noOfSeeds = Gossiper.instance.seeds.size();
final AtomicBoolean ackSend = new AtomicBoolean(false);
MockMessagingSpy spySyn = MockMessagingService.when(verb(MessagingService.Verb.GOSSIP_DIGEST_SYN))
.respondN((msgOut, to) ->
{
// ACK once to finish shadow round, then busy-spin until gossiper has been enabled
// and then reply with remaining ACKs from other seeds
if (!ackSend.compareAndSet(false, true))
{
while (!Gossiper.instance.isEnabled()) ;
}
HeartBeatState hb = new HeartBeatState(123, 456);
EndpointState state = new EndpointState(hb);
GossipDigestAck payload = new GossipDigestAck(
Collections.singletonList(new GossipDigest(to, hb.getGeneration(), hb.getHeartBeatVersion())),
Collections.singletonMap(to, state));
logger.debug("Simulating digest ACK reply");
return MessageIn.create(to, payload, Collections.emptyMap(), MessagingService.Verb.GOSSIP_DIGEST_ACK, MessagingService.current_version);
}, noOfSeeds);
// GossipDigestAckVerbHandler will send ack2 for each ack received (after the shadow round)
MockMessagingSpy spyAck2 = MockMessagingService.when(verb(MessagingService.Verb.GOSSIP_DIGEST_ACK2)).dontReply();
// Migration request messages should not be emitted during shadow round
MockMessagingSpy spyMigrationReq = MockMessagingService.when(verb(MessagingService.Verb.MIGRATION_REQUEST)).dontReply();
try
{
StorageService.instance.initServer();
}
catch (Exception e)
{
assertEquals("Unable to contact any seeds!", e.getMessage());
}
// we expect one SYN for each seed during shadow round + additional SYNs after gossiper has been enabled
assertTrue(spySyn.messagesIntercepted > noOfSeeds);
// we don't expect to emit any GOSSIP_DIGEST_ACK2 or MIGRATION_REQUEST messages
assertEquals(0, spyAck2.messagesIntercepted);
assertEquals(0, spyMigrationReq.messagesIntercepted);
}
}

24
test/unit/org/apache/cassandra/net/MatcherResponse.java
View File

@ -173,16 +173,22 @@ public class MatcherResponse
assert !sendResponses.contains(id) : "ID re-use for outgoing message";
sendResponses.add(id);
}
MessageIn<?> response = fnResponse.apply(message, to);
if (response != null)
// create response asynchronously to match request/response communication execution behavior
new Thread(() ->
{
CallbackInfo cb = MessagingService.instance().getRegisteredCallback(id);
if (cb != null)
cb.callback.response(response);
else
MessagingService.instance().receive(response, id);
spy.matchingResponse(response);
}
MessageIn<?> response = fnResponse.apply(message, to);
if (response != null)
{
CallbackInfo cb = MessagingService.instance().getRegisteredCallback(id);
if (cb != null)
cb.callback.response(response);
else
MessagingService.instance().receive(response, id);
spy.matchingResponse(response);
}
}).start();
return false;
}
return true;