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

Properly handle hinted handoff after topology changes 

patch by Branimir Lambov; reviewed by Aleksey Yeschenko for
CASSANDRA-5902
This commit is contained in:
Branimir Lambov 2015-11-18 16:26:37 +02:00 committed by Aleksey Yeschenko
parent 2a824c024a
commit b2f38ef177
9 changed files with 199 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
3.0.4
* Properly handle hinted handoff after topology changes (CASSANDRA-5902)
* AssertionError when listing sstable files on inconsistent disk state (CASSANDRA-11156)
* Fix wrong rack counting and invalid conditions check for TokenAllocation
(CASSANDRA-11139)

3
src/java/org/apache/cassandra/batchlog/BatchlogManager.java
View File

@ -420,8 +420,7 @@ public class BatchlogManager implements BatchlogManagerMBean
String ks = mutation.getKeyspaceName();
Token tk = mutation.key().getToken();
for (InetAddress endpoint : Iterables.concat(StorageService.instance.getNaturalEndpoints(ks, tk),
StorageService.instance.getTokenMetadata().pendingEndpointsFor(tk, ks)))
for (InetAddress endpoint : StorageService.instance.getNaturalAndPendingEndpoints(ks, tk))
{
if (endpoint.equals(FBUtilities.getBroadcastAddress()))
{

22
src/java/org/apache/cassandra/hints/HintVerbHandler.java
View File

@ -29,6 +29,7 @@ import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.MessageIn;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.serializers.MarshalException;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.service.StorageService;
/**
@ -72,12 +73,23 @@ public final class HintVerbHandler implements IVerbHandler<HintMessage>
return;
}
// Apply the hint if this node is the destination, store for future dispatch if this node isn't (must have gotten
// it from a decommissioned node that had streamed it before going out).
if (hostId.equals(StorageService.instance.getLocalHostUUID()))
hint.apply();
else
if (!hostId.equals(StorageService.instance.getLocalHostUUID()))
{
// the node is not the final destination of the hint (must have gotten it from a decommissioning node),
// so just store it locally, to be delivered later.
HintsService.instance.write(hostId, hint);
}
else if (!StorageProxy.instance.appliesLocally(hint.mutation))
{
// the topology has changed, and we are no longer a replica of the mutation - since we don't know which node(s)
// it has been handed over to, re-address the hint to all replicas; see CASSANDRA-5902.
HintsService.instance.writeForAllReplicas(hint);
}
else
{
// the common path - the node is both the destination and a valid replica for the hint.
hint.apply();
}
reply(id, message.from);
}

28
src/java/org/apache/cassandra/hints/HintsDispatchExecutor.java
View File

@ -18,6 +18,7 @@
package org.apache.cassandra.hints;
import java.io.File;
import java.net.InetAddress;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.*;
@ -236,10 +237,21 @@ final class HintsDispatchExecutor
{
logger.trace("Dispatching hints file {}", descriptor.fileName());
InetAddress address = StorageService.instance.getEndpointForHostId(hostId);
if (address != null)
return deliver(descriptor, address);
// address == null means the target no longer exist; find new home for each hint entry.
convert(descriptor);
return true;
}
private boolean deliver(HintsDescriptor descriptor, InetAddress address)
{
File file = new File(hintsDirectory, descriptor.fileName());
Long offset = store.getDispatchOffset(descriptor).orElse(null);
try (HintsDispatcher dispatcher = HintsDispatcher.create(file, rateLimiter, hostId, descriptor.hostId, isPaused))
try (HintsDispatcher dispatcher = HintsDispatcher.create(file, rateLimiter, address, descriptor.hostId, isPaused))
{
if (offset != null)
dispatcher.seek(offset);
@ -260,5 +272,19 @@ final class HintsDispatchExecutor
}
}
}
// for each hint in the hints file for a node that isn't part of the ring anymore, write RF hints for each replica
private void convert(HintsDescriptor descriptor)
{
File file = new File(hintsDirectory, descriptor.fileName());
try (HintsReader reader = HintsReader.open(file, rateLimiter))
{
reader.forEach(page -> page.hintsIterator().forEachRemaining(HintsService.instance::writeForAllReplicas));
store.delete(descriptor);
store.cleanUp(descriptor);
logger.info("Finished converting hints file {}", descriptor.fileName());
}
}
}
}

7
src/java/org/apache/cassandra/hints/HintsDispatcher.java
View File

@ -32,7 +32,6 @@ import org.apache.cassandra.gms.FailureDetector;
import org.apache.cassandra.net.IAsyncCallbackWithFailure;
import org.apache.cassandra.net.MessageIn;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
/**
@ -64,11 +63,10 @@ final class HintsDispatcher implements AutoCloseable
this.isPaused = isPaused;
}
static HintsDispatcher create(File file, RateLimiter rateLimiter, UUID hostId, UUID hintFor, AtomicBoolean isPaused)
static HintsDispatcher create(File file, RateLimiter rateLimiter, InetAddress address, UUID hostId, AtomicBoolean isPaused)
{
InetAddress address = StorageService.instance.getEndpointForHostId(hostId);
int messagingVersion = MessagingService.instance().getVersion(address);
return new HintsDispatcher(HintsReader.open(file, rateLimiter), hintFor, address, messagingVersion, isPaused);
return new HintsDispatcher(HintsReader.open(file, rateLimiter), hostId, address, messagingVersion, isPaused);
}
public void close()
@ -152,6 +150,7 @@ final class HintsDispatcher implements AutoCloseable
/*
* Sending hints in compatibility mode.
*/
private <T> Action sendHints(Iterator<T> hints, Collection<Callback> callbacks, Function<T, Callback> sendFunction)
{
while (hints.hasNext())

20
src/java/org/apache/cassandra/hints/HintsService.java
View File

@ -22,7 +22,6 @@ import java.lang.management.ManagementFactory;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Collections;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
@ -40,9 +39,11 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.ParameterizedClass;
import org.apache.cassandra.metrics.HintedHandoffMetrics;
import org.apache.cassandra.metrics.StorageMetrics;
import org.apache.cassandra.schema.CompressionParams;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.service.StorageService;
import static com.google.common.collect.Iterables.filter;
import static com.google.common.collect.Iterables.transform;
import static com.google.common.collect.Iterables.size;
@ -166,6 +167,21 @@ public final class HintsService implements HintsServiceMBean
write(Collections.singleton(hostId), hint);
}
/**
* Write a hint for all replicas. Used to re-dispatch hints whose destination is either missing or no longer correct.
*/
void writeForAllReplicas(Hint hint)
{
String keyspaceName = hint.mutation.getKeyspaceName();
Token token = hint.mutation.key().getToken();
Iterable<UUID> hostIds =
transform(filter(StorageService.instance.getNaturalAndPendingEndpoints(keyspaceName, token), StorageProxy::shouldHint),
StorageService.instance::getHostIdForEndpoint);
write(hostIds, hint);
}
/**
* Flush the buffer pool for the selected target nodes, then fsync their writers.
*

20
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -71,6 +71,8 @@ import org.apache.cassandra.triggers.TriggerExecutor;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.utils.AbstractIterator;
import static com.google.common.collect.Iterables.contains;
public class StorageProxy implements StorageProxyMBean
{
public static final String MBEAN_NAME = "org.apache.cassandra.db:type=StorageProxy";
@ -670,12 +672,10 @@ public class StorageProxy implements StorageProxyMBean
private static void hintMutation(Mutation mutation)
{
Token tk = DatabaseDescriptor.getPartitioner().getToken(mutation.key().getKey());
List<InetAddress> naturalEndpoints = StorageService.instance.getNaturalEndpoints(mutation.getKeyspaceName(), tk);
Collection<InetAddress> pendingEndpoints =
StorageService.instance.getTokenMetadata().pendingEndpointsFor(tk, mutation.getKeyspaceName());
String keyspaceName = mutation.getKeyspaceName();
Token token = mutation.key().getToken();
Iterable<InetAddress> endpoints = Iterables.concat(naturalEndpoints, pendingEndpoints);
Iterable<InetAddress> endpoints = StorageService.instance.getNaturalAndPendingEndpoints(keyspaceName, token);
ArrayList<InetAddress> endpointsToHint = new ArrayList<>(Iterables.size(endpoints));
// local writes can timeout, but cannot be dropped (see LocalMutationRunnable and CASSANDRA-6510),
@ -687,6 +687,16 @@ public class StorageProxy implements StorageProxyMBean
submitHint(mutation, endpointsToHint, null);
}
public boolean appliesLocally(Mutation mutation)
{
String keyspaceName = mutation.getKeyspaceName();
Token token = mutation.key().getToken();
InetAddress local = FBUtilities.getBroadcastAddress();
return StorageService.instance.getNaturalEndpoints(keyspaceName, token).contains(local)
|| StorageService.instance.getTokenMetadata().pendingEndpointsFor(token, keyspaceName).contains(local);
}
/**
* Use this method to have these Mutations applied
* across all replicas.

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

@ -3300,6 +3300,14 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
return Keyspace.open(keyspaceName).getReplicationStrategy().getNaturalEndpoints(pos);
}
/**
* Returns the endpoints currently responsible for storing the token plus pending ones
*/
public Iterable<InetAddress> getNaturalAndPendingEndpoints(String keyspaceName, Token token)
{
return Iterables.concat(getNaturalEndpoints(keyspaceName, token), tokenMetadata.pendingEndpointsFor(token, keyspaceName));
}
/**
* This method attempts to return N endpoints that are responsible for storing the
* specified key i.e for replication.

109
test/unit/org/apache/cassandra/hints/HintTest.java
View File

@ -18,6 +18,11 @@
package org.apache.cassandra.hints;
import java.io.IOException;
import java.net.InetAddress;
import java.util.Collections;
import java.util.UUID;
import com.google.common.collect.ImmutableList;
import org.junit.Before;
import org.junit.BeforeClass;
@ -26,17 +31,24 @@ import org.junit.Test;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.Util;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.partitions.FilteredPartition;
import org.apache.cassandra.db.partitions.PartitionIterator;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.dht.BootStrapper;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.metrics.StorageMetrics;
import org.apache.cassandra.net.MessageIn;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.TableParams;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.FBUtilities;
import static junit.framework.Assert.*;
@ -67,6 +79,12 @@ public class HintTest
@Before
public void resetGcGraceSeconds()
{
TokenMetadata tokenMeta = StorageService.instance.getTokenMetadata();
InetAddress local = FBUtilities.getBroadcastAddress();
tokenMeta.clearUnsafe();
tokenMeta.updateHostId(UUID.randomUUID(), local);
tokenMeta.updateNormalTokens(BootStrapper.getRandomTokens(tokenMeta, 1), local);
for (CFMetaData table : Schema.instance.getTablesAndViews(KEYSPACE))
table.gcGraceSeconds(TableParams.DEFAULT_GC_GRACE_SECONDS);
}
@ -185,6 +203,97 @@ public class HintTest
assertNoPartitions(key, TABLE2);
}
@SuppressWarnings("unchecked")
@Test
public void testChangedTopology() throws Exception
{
// create a hint
long now = FBUtilities.timestampMicros();
String key = "testChangedTopology";
Mutation mutation = createMutation(key, now);
Hint hint = Hint.create(mutation, now / 1000);
// Prepare metadata with injected stale endpoint serving the mutation key.
TokenMetadata tokenMeta = StorageService.instance.getTokenMetadata();
InetAddress local = FBUtilities.getBroadcastAddress();
InetAddress endpoint = InetAddress.getByName("1.1.1.1");
UUID localId = StorageService.instance.getLocalHostUUID();
UUID targetId = UUID.randomUUID();
tokenMeta.updateHostId(targetId, endpoint);
tokenMeta.updateNormalTokens(ImmutableList.of(mutation.key().getToken()), endpoint);
// sanity check that there is no data inside yet
assertNoPartitions(key, TABLE0);
assertNoPartitions(key, TABLE1);
assertNoPartitions(key, TABLE2);
assert StorageProxy.instance.getHintsInProgress() == 0;
long totalHintCount = StorageProxy.instance.getTotalHints();
// Process hint message.
HintMessage message = new HintMessage(localId, hint);
MessagingService.instance().getVerbHandler(MessagingService.Verb.HINT).doVerb(
MessageIn.create(local, message, Collections.emptyMap(), MessagingService.Verb.HINT, MessagingService.current_version),
-1);
// hint should not be applied as we no longer are a replica
assertNoPartitions(key, TABLE0);
assertNoPartitions(key, TABLE1);
assertNoPartitions(key, TABLE2);
// Attempt to send to new endpoint should have been made. Node is not live hence it should now be a hint.
assertEquals(totalHintCount + 1, StorageProxy.instance.getTotalHints());
}
@SuppressWarnings("unchecked")
@Test
public void testChangedTopologyNotHintable() throws Exception
{
// create a hint
long now = FBUtilities.timestampMicros();
String key = "testChangedTopology";
Mutation mutation = createMutation(key, now);
Hint hint = Hint.create(mutation, now / 1000);
// Prepare metadata with injected stale endpoint.
TokenMetadata tokenMeta = StorageService.instance.getTokenMetadata();
InetAddress local = FBUtilities.getBroadcastAddress();
InetAddress endpoint = InetAddress.getByName("1.1.1.1");
UUID localId = StorageService.instance.getLocalHostUUID();
UUID targetId = UUID.randomUUID();
tokenMeta.updateHostId(targetId, endpoint);
tokenMeta.updateNormalTokens(ImmutableList.of(mutation.key().getToken()), endpoint);
// sanity check that there is no data inside yet
assertNoPartitions(key, TABLE0);
assertNoPartitions(key, TABLE1);
assertNoPartitions(key, TABLE2);
try
{
DatabaseDescriptor.setHintedHandoffEnabled(false);
assert StorageMetrics.totalHintsInProgress.getCount() == 0;
long totalHintCount = StorageMetrics.totalHints.getCount();
// Process hint message.
HintMessage message = new HintMessage(localId, hint);
MessagingService.instance().getVerbHandler(MessagingService.Verb.HINT).doVerb(
MessageIn.create(local, message, Collections.emptyMap(), MessagingService.Verb.HINT, MessagingService.current_version),
-1);
// hint should not be applied as we no longer are a replica
assertNoPartitions(key, TABLE0);
assertNoPartitions(key, TABLE1);
assertNoPartitions(key, TABLE2);
// Attempt to send to new endpoint should not have been made.
assertEquals(totalHintCount, StorageMetrics.totalHints.getCount());
}
finally
{
DatabaseDescriptor.setHintedHandoffEnabled(true);
}
}
private static Mutation createMutation(String key, long now)
{
Mutation mutation = new Mutation(KEYSPACE, dk(key));