mirror of https://github.com/apache/cassandra
Merge branch 'cassandra-2.1' into cassandra-2.2
This commit is contained in:
Robert Stupp
commit
432a8a4843
|
|
@ -16,6 +16,7 @@
|
|||
* Deprecate Pig support (CASSANDRA-10542)
|
||||
* Reduce contention getting instances of CompositeType (CASSANDRA-10433)
|
||||
Merged from 2.1:
|
||||
* Warn or fail when changing cluster topology live (CASSANDRA-10243)
|
||||
* Status command in debian/ubuntu init script doesn't work (CASSANDRA-10213)
|
||||
* Some DROP ... IF EXISTS incorrectly result in exceptions on non-existing KS (CASSANDRA-10658)
|
||||
* DeletionTime.compareTo wrong in rare cases (CASSANDRA-10749)
|
||||
|
|
|
|||
6
NEWS.txt
6
NEWS.txt
|
|
@ -40,6 +40,12 @@ New features
|
|||
- Native protocol server now allows both SSL and non-SSL connections on
|
||||
the same port.
|
||||
|
||||
Operations
|
||||
------------
|
||||
- Changing rack or dc of live nodes is no longer possible for PropertyFileSnitch
|
||||
and YamlFileNetworkTopologySnitch. Reloading the configuration file of
|
||||
GossipingPropertyFileSnitch has been disabled, CASSANDRA-10243.
|
||||
|
||||
|
||||
2.2.3
|
||||
=====
|
||||
|
|
|
|||
|
|
@ -256,24 +256,12 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
|
|||
subscribers.remove(subscriber);
|
||||
}
|
||||
|
||||
public Set<InetAddress> getLiveMembers()
|
||||
public Set<InetAddress> getLiveEndpoints()
|
||||
{
|
||||
Set<InetAddress> liveMembers = new HashSet<InetAddress>(liveEndpoints);
|
||||
if (!liveMembers.contains(FBUtilities.getBroadcastAddress()))
|
||||
liveMembers.add(FBUtilities.getBroadcastAddress());
|
||||
return liveMembers;
|
||||
}
|
||||
|
||||
public Set<InetAddress> getLiveTokenOwners()
|
||||
{
|
||||
Set<InetAddress> tokenOwners = new HashSet<InetAddress>();
|
||||
for (InetAddress member : getLiveMembers())
|
||||
{
|
||||
EndpointState epState = endpointStateMap.get(member);
|
||||
if (epState != null && !isDeadState(epState) && StorageService.instance.getTokenMetadata().isMember(member))
|
||||
tokenOwners.add(member);
|
||||
}
|
||||
return tokenOwners;
|
||||
Set<InetAddress> liveEndpoints = new HashSet<InetAddress>(this.liveEndpoints);
|
||||
if (!liveEndpoints.contains(FBUtilities.getBroadcastAddress()))
|
||||
liveEndpoints.add(FBUtilities.getBroadcastAddress());
|
||||
return liveEndpoints;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -967,7 +955,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
|
|||
MessagingService.instance().sendRR(echoMessage, addr, echoHandler);
|
||||
}
|
||||
|
||||
private void realMarkAlive(final InetAddress addr, final EndpointState localState)
|
||||
@VisibleForTesting
|
||||
public void realMarkAlive(final InetAddress addr, final EndpointState localState)
|
||||
{
|
||||
if (logger.isTraceEnabled())
|
||||
logger.trace("marking as alive {}", addr);
|
||||
|
|
@ -984,7 +973,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
|
|||
logger.trace("Notified {}", subscribers);
|
||||
}
|
||||
|
||||
private void markDead(InetAddress addr, EndpointState localState)
|
||||
@VisibleForTesting
|
||||
public void markDead(InetAddress addr, EndpointState localState)
|
||||
{
|
||||
if (logger.isTraceEnabled())
|
||||
logger.trace("marking as down {}", addr);
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
/**
|
||||
/*
|
||||
* 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
|
||||
|
|
@ -32,8 +32,6 @@ import org.apache.cassandra.gms.EndpointState;
|
|||
import org.apache.cassandra.gms.Gossiper;
|
||||
import org.apache.cassandra.service.StorageService;
|
||||
import org.apache.cassandra.utils.FBUtilities;
|
||||
import org.apache.cassandra.utils.ResourceWatcher;
|
||||
import org.apache.cassandra.utils.WrappedRunnable;
|
||||
|
||||
|
||||
public class GossipingPropertyFileSnitch extends AbstractNetworkTopologySnitch// implements IEndpointStateChangeSubscriber
|
||||
|
|
@ -42,28 +40,23 @@ public class GossipingPropertyFileSnitch extends AbstractNetworkTopologySnitch//
|
|||
|
||||
private PropertyFileSnitch psnitch;
|
||||
|
||||
private volatile String myDC;
|
||||
private volatile String myRack;
|
||||
private volatile boolean preferLocal;
|
||||
private AtomicReference<ReconnectableSnitchHelper> snitchHelperReference;
|
||||
private volatile boolean gossipStarted;
|
||||
private final String myDC;
|
||||
private final String myRack;
|
||||
private final boolean preferLocal;
|
||||
private final AtomicReference<ReconnectableSnitchHelper> snitchHelperReference;
|
||||
|
||||
private Map<InetAddress, Map<String, String>> savedEndpoints;
|
||||
private static final String DEFAULT_DC = "UNKNOWN_DC";
|
||||
private static final String DEFAULT_RACK = "UNKNOWN_RACK";
|
||||
|
||||
private static final int DEFAULT_REFRESH_PERIOD_IN_SECONDS = 60;
|
||||
|
||||
public GossipingPropertyFileSnitch() throws ConfigurationException
|
||||
{
|
||||
this(DEFAULT_REFRESH_PERIOD_IN_SECONDS);
|
||||
}
|
||||
SnitchProperties properties = loadConfiguration();
|
||||
|
||||
public GossipingPropertyFileSnitch(int refreshPeriodInSeconds) throws ConfigurationException
|
||||
{
|
||||
snitchHelperReference = new AtomicReference<ReconnectableSnitchHelper>();
|
||||
|
||||
reloadConfiguration(false);
|
||||
myDC = properties.get("dc", DEFAULT_DC).trim();
|
||||
myRack = properties.get("rack", DEFAULT_RACK).trim();
|
||||
preferLocal = Boolean.parseBoolean(properties.get("prefer_local", "false"));
|
||||
snitchHelperReference = new AtomicReference<>();
|
||||
|
||||
try
|
||||
{
|
||||
|
|
@ -74,23 +67,15 @@ public class GossipingPropertyFileSnitch extends AbstractNetworkTopologySnitch//
|
|||
{
|
||||
logger.info("Unable to load {}; compatibility mode disabled", PropertyFileSnitch.SNITCH_PROPERTIES_FILENAME);
|
||||
}
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
FBUtilities.resourceToFile(SnitchProperties.RACKDC_PROPERTY_FILENAME);
|
||||
Runnable runnable = new WrappedRunnable()
|
||||
{
|
||||
protected void runMayThrow() throws ConfigurationException
|
||||
{
|
||||
reloadConfiguration(true);
|
||||
}
|
||||
};
|
||||
ResourceWatcher.watch(SnitchProperties.RACKDC_PROPERTY_FILENAME, runnable, refreshPeriodInSeconds * 1000);
|
||||
}
|
||||
catch (ConfigurationException ex)
|
||||
{
|
||||
logger.error("{} found, but does not look like a plain file. Will not watch it for changes", SnitchProperties.RACKDC_PROPERTY_FILENAME);
|
||||
}
|
||||
private static SnitchProperties loadConfiguration() throws ConfigurationException
|
||||
{
|
||||
final SnitchProperties properties = new SnitchProperties();
|
||||
if (!properties.contains("dc") || !properties.contains("rack"))
|
||||
throw new ConfigurationException("DC or rack not found in snitch properties, check your configuration in: " + SnitchProperties.RACKDC_PROPERTY_FILENAME);
|
||||
|
||||
return properties;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -156,56 +141,18 @@ public class GossipingPropertyFileSnitch extends AbstractNetworkTopologySnitch//
|
|||
Gossiper.instance.addLocalApplicationState(ApplicationState.INTERNAL_IP,
|
||||
StorageService.instance.valueFactory.internalIP(FBUtilities.getLocalAddress().getHostAddress()));
|
||||
|
||||
reloadGossiperState();
|
||||
|
||||
gossipStarted = true;
|
||||
}
|
||||
|
||||
private void reloadConfiguration(boolean isUpdate) throws ConfigurationException
|
||||
{
|
||||
final SnitchProperties properties = new SnitchProperties();
|
||||
|
||||
String newDc = properties.get("dc", null);
|
||||
String newRack = properties.get("rack", null);
|
||||
if (newDc == null || newRack == null)
|
||||
throw new ConfigurationException("DC or rack not found in snitch properties, check your configuration in: " + SnitchProperties.RACKDC_PROPERTY_FILENAME);
|
||||
|
||||
newDc = newDc.trim();
|
||||
newRack = newRack.trim();
|
||||
final boolean newPreferLocal = Boolean.parseBoolean(properties.get("prefer_local", "false"));
|
||||
|
||||
if (!newDc.equals(myDC) || !newRack.equals(myRack) || (preferLocal != newPreferLocal))
|
||||
{
|
||||
myDC = newDc;
|
||||
myRack = newRack;
|
||||
preferLocal = newPreferLocal;
|
||||
|
||||
reloadGossiperState();
|
||||
|
||||
if (StorageService.instance != null)
|
||||
{
|
||||
if (isUpdate)
|
||||
StorageService.instance.updateTopology(FBUtilities.getBroadcastAddress());
|
||||
else
|
||||
StorageService.instance.getTokenMetadata().invalidateCachedRings();
|
||||
}
|
||||
|
||||
if (gossipStarted)
|
||||
StorageService.instance.gossipSnitchInfo();
|
||||
}
|
||||
loadGossiperState();
|
||||
}
|
||||
|
||||
private void reloadGossiperState()
|
||||
private void loadGossiperState()
|
||||
{
|
||||
if (Gossiper.instance != null)
|
||||
{
|
||||
ReconnectableSnitchHelper pendingHelper = new ReconnectableSnitchHelper(this, myDC, preferLocal);
|
||||
Gossiper.instance.register(pendingHelper);
|
||||
|
||||
pendingHelper = snitchHelperReference.getAndSet(pendingHelper);
|
||||
if (pendingHelper != null)
|
||||
Gossiper.instance.unregister(pendingHelper);
|
||||
}
|
||||
// else this will eventually rerun at gossiperStarting()
|
||||
assert Gossiper.instance != null;
|
||||
|
||||
ReconnectableSnitchHelper pendingHelper = new ReconnectableSnitchHelper(this, myDC, preferLocal);
|
||||
Gossiper.instance.register(pendingHelper);
|
||||
|
||||
pendingHelper = snitchHelperReference.getAndSet(pendingHelper);
|
||||
if (pendingHelper != null)
|
||||
Gossiper.instance.unregister(pendingHelper);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -23,12 +23,16 @@ import java.net.UnknownHostException;
|
|||
import java.util.Arrays;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import java.util.Properties;
|
||||
import java.util.Set;
|
||||
|
||||
import com.google.common.collect.Sets;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import org.apache.cassandra.exceptions.ConfigurationException;
|
||||
import org.apache.cassandra.gms.Gossiper;
|
||||
import org.apache.cassandra.io.util.FileUtils;
|
||||
import org.apache.cassandra.service.StorageService;
|
||||
import org.apache.cassandra.utils.FBUtilities;
|
||||
|
|
@ -52,6 +56,7 @@ public class PropertyFileSnitch extends AbstractNetworkTopologySnitch
|
|||
private static final Logger logger = LoggerFactory.getLogger(PropertyFileSnitch.class);
|
||||
|
||||
public static final String SNITCH_PROPERTIES_FILENAME = "cassandra-topology.properties";
|
||||
private static final int DEFAULT_REFRESH_PERIOD_IN_SECONDS = 5;
|
||||
|
||||
private static volatile Map<InetAddress, String[]> endpointMap;
|
||||
private static volatile String[] defaultDCRack;
|
||||
|
|
@ -59,6 +64,11 @@ public class PropertyFileSnitch extends AbstractNetworkTopologySnitch
|
|||
private volatile boolean gossipStarted;
|
||||
|
||||
public PropertyFileSnitch() throws ConfigurationException
|
||||
{
|
||||
this(DEFAULT_REFRESH_PERIOD_IN_SECONDS);
|
||||
}
|
||||
|
||||
public PropertyFileSnitch(int refreshPeriodInSeconds) throws ConfigurationException
|
||||
{
|
||||
reloadConfiguration(false);
|
||||
|
||||
|
|
@ -72,7 +82,7 @@ public class PropertyFileSnitch extends AbstractNetworkTopologySnitch
|
|||
reloadConfiguration(true);
|
||||
}
|
||||
};
|
||||
ResourceWatcher.watch(SNITCH_PROPERTIES_FILENAME, runnable, 60 * 1000);
|
||||
ResourceWatcher.watch(SNITCH_PROPERTIES_FILENAME, runnable, refreshPeriodInSeconds * 1000);
|
||||
}
|
||||
catch (ConfigurationException ex)
|
||||
{
|
||||
|
|
@ -86,7 +96,7 @@ public class PropertyFileSnitch extends AbstractNetworkTopologySnitch
|
|||
* @param endpoint endpoint to process
|
||||
* @return a array of string with the first index being the data center and the second being the rack
|
||||
*/
|
||||
public String[] getEndpointInfo(InetAddress endpoint)
|
||||
public static String[] getEndpointInfo(InetAddress endpoint)
|
||||
{
|
||||
String[] rawEndpointInfo = getRawEndpointInfo(endpoint);
|
||||
if (rawEndpointInfo == null)
|
||||
|
|
@ -94,7 +104,7 @@ public class PropertyFileSnitch extends AbstractNetworkTopologySnitch
|
|||
return rawEndpointInfo;
|
||||
}
|
||||
|
||||
private String[] getRawEndpointInfo(InetAddress endpoint)
|
||||
private static String[] getRawEndpointInfo(InetAddress endpoint)
|
||||
{
|
||||
String[] value = endpointMap.get(endpoint);
|
||||
if (value == null)
|
||||
|
|
@ -133,7 +143,8 @@ public class PropertyFileSnitch extends AbstractNetworkTopologySnitch
|
|||
|
||||
public void reloadConfiguration(boolean isUpdate) throws ConfigurationException
|
||||
{
|
||||
HashMap<InetAddress, String[]> reloadedMap = new HashMap<InetAddress, String[]>();
|
||||
HashMap<InetAddress, String[]> reloadedMap = new HashMap<>();
|
||||
String[] reloadedDefaultDCRack = null;
|
||||
|
||||
Properties properties = new Properties();
|
||||
try (InputStream stream = getClass().getClassLoader().getResourceAsStream(SNITCH_PROPERTIES_FILENAME))
|
||||
|
|
@ -150,18 +161,18 @@ public class PropertyFileSnitch extends AbstractNetworkTopologySnitch
|
|||
String key = (String) entry.getKey();
|
||||
String value = (String) entry.getValue();
|
||||
|
||||
if (key.equals("default"))
|
||||
if ("default".equals(key))
|
||||
{
|
||||
String[] newDefault = value.split(":");
|
||||
if (newDefault.length < 2)
|
||||
defaultDCRack = new String[] { "default", "default" };
|
||||
reloadedDefaultDCRack = new String[] { "default", "default" };
|
||||
else
|
||||
defaultDCRack = new String[] { newDefault[0].trim(), newDefault[1].trim() };
|
||||
reloadedDefaultDCRack = new String[] { newDefault[0].trim(), newDefault[1].trim() };
|
||||
}
|
||||
else
|
||||
{
|
||||
InetAddress host;
|
||||
String hostString = key.replace("/", "");
|
||||
String hostString = StringUtils.remove(key, '/');
|
||||
try
|
||||
{
|
||||
host = InetAddress.getByName(hostString);
|
||||
|
|
@ -178,18 +189,24 @@ public class PropertyFileSnitch extends AbstractNetworkTopologySnitch
|
|||
reloadedMap.put(host, token);
|
||||
}
|
||||
}
|
||||
if (defaultDCRack == null && !reloadedMap.containsKey(FBUtilities.getBroadcastAddress()))
|
||||
throw new ConfigurationException(String.format("Snitch definitions at %s do not define a location for this node's broadcast address %s, nor does it provides a default",
|
||||
if (reloadedDefaultDCRack == null && !reloadedMap.containsKey(FBUtilities.getBroadcastAddress()))
|
||||
throw new ConfigurationException(String.format("Snitch definitions at %s do not define a location for " +
|
||||
"this node's broadcast address %s, nor does it provides a default",
|
||||
SNITCH_PROPERTIES_FILENAME, FBUtilities.getBroadcastAddress()));
|
||||
|
||||
if (isUpdate && !livenessCheck(reloadedMap, reloadedDefaultDCRack))
|
||||
return;
|
||||
|
||||
if (logger.isTraceEnabled())
|
||||
{
|
||||
StringBuilder sb = new StringBuilder();
|
||||
for (Map.Entry<InetAddress, String[]> entry : reloadedMap.entrySet())
|
||||
sb.append(entry.getKey()).append(":").append(Arrays.toString(entry.getValue())).append(", ");
|
||||
sb.append(entry.getKey()).append(':').append(Arrays.toString(entry.getValue())).append(", ");
|
||||
logger.trace("Loaded network topology from property file: {}", StringUtils.removeEnd(sb.toString(), ", "));
|
||||
}
|
||||
|
||||
|
||||
defaultDCRack = reloadedDefaultDCRack;
|
||||
endpointMap = reloadedMap;
|
||||
if (StorageService.instance != null) // null check tolerates circular dependency; see CASSANDRA-4145
|
||||
{
|
||||
|
|
@ -203,6 +220,41 @@ public class PropertyFileSnitch extends AbstractNetworkTopologySnitch
|
|||
StorageService.instance.gossipSnitchInfo();
|
||||
}
|
||||
|
||||
/**
|
||||
* We cannot update rack or data-center for a live node, see CASSANDRA-10243.
|
||||
*
|
||||
* @param reloadedMap - the new map of hosts to dc:rack properties
|
||||
* @param reloadedDefaultDCRack - the default dc:rack or null if no default
|
||||
* @return true if we can continue updating (no live host had dc or rack updated)
|
||||
*/
|
||||
private static boolean livenessCheck(HashMap<InetAddress, String[]> reloadedMap, String[] reloadedDefaultDCRack)
|
||||
{
|
||||
// If the default has changed we must check all live hosts but hopefully we will find a live
|
||||
// host quickly and interrupt the loop. Otherwise we only check the live hosts that were either
|
||||
// in the old set or in the new set
|
||||
Set<InetAddress> hosts = Arrays.equals(defaultDCRack, reloadedDefaultDCRack)
|
||||
? Sets.intersection(StorageService.instance.getLiveMembers(), // same default
|
||||
Sets.union(endpointMap.keySet(), reloadedMap.keySet()))
|
||||
: StorageService.instance.getLiveMembers(); // default updated
|
||||
|
||||
for (InetAddress host : hosts)
|
||||
{
|
||||
String[] origValue = endpointMap.containsKey(host) ? endpointMap.get(host) : defaultDCRack;
|
||||
String[] updateValue = reloadedMap.containsKey(host) ? reloadedMap.get(host) : reloadedDefaultDCRack;
|
||||
|
||||
if (!Arrays.equals(origValue, updateValue))
|
||||
{
|
||||
logger.error("Cannot update data center or rack from {} to {} for live host {}, property file NOT RELOADED",
|
||||
origValue,
|
||||
updateValue,
|
||||
host);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void gossiperStarting()
|
||||
{
|
||||
|
|
|
|||
|
|
@ -66,4 +66,9 @@ public class SnitchProperties
|
|||
{
|
||||
return properties.getProperty(propertyName, defaultValue);
|
||||
}
|
||||
|
||||
public boolean contains(String propertyName)
|
||||
{
|
||||
return properties.containsKey(propertyName);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -232,15 +232,18 @@ public class CassandraDaemon
|
|||
|
||||
Keyspace.setInitialized();
|
||||
|
||||
String storedRack = SystemKeyspace.getRack();
|
||||
if (storedRack != null)
|
||||
if (!Boolean.getBoolean("cassandra.ignore_rack"))
|
||||
{
|
||||
String currentRack = DatabaseDescriptor.getEndpointSnitch().getRack(FBUtilities.getBroadcastAddress());
|
||||
if (!storedRack.equals(currentRack))
|
||||
String storedRack = SystemKeyspace.getRack();
|
||||
if (storedRack != null)
|
||||
{
|
||||
logger.error("Cannot start node if snitch's rack differs from previous rack. " +
|
||||
"Please fix the snitch or decommission and rebootstrap this node.");
|
||||
System.exit(100);
|
||||
String currentRack = DatabaseDescriptor.getEndpointSnitch().getRack(FBUtilities.getBroadcastAddress());
|
||||
if (!storedRack.equals(currentRack))
|
||||
{
|
||||
logger.error("Cannot start node if snitch's rack differs from previous rack. " +
|
||||
"Please fix the snitch or decommission and rebootstrap this node.");
|
||||
System.exit(100);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -469,7 +469,7 @@ public class MigrationManager
|
|||
}
|
||||
});
|
||||
|
||||
for (InetAddress endpoint : Gossiper.instance.getLiveMembers())
|
||||
for (InetAddress endpoint : Gossiper.instance.getLiveEndpoints())
|
||||
{
|
||||
// only push schema to nodes with known and equal versions
|
||||
if (!endpoint.equals(FBUtilities.getBroadcastAddress()) &&
|
||||
|
|
@ -511,7 +511,7 @@ public class MigrationManager
|
|||
|
||||
Schema.instance.clear();
|
||||
|
||||
Set<InetAddress> liveEndpoints = Gossiper.instance.getLiveMembers();
|
||||
Set<InetAddress> liveEndpoints = Gossiper.instance.getLiveEndpoints();
|
||||
liveEndpoints.remove(FBUtilities.getBroadcastAddress());
|
||||
|
||||
// force migration if there are nodes around
|
||||
|
|
|
|||
|
|
@ -1929,7 +1929,7 @@ public class StorageProxy implements StorageProxyMBean
|
|||
{
|
||||
final String myVersion = Schema.instance.getVersion().toString();
|
||||
final Map<InetAddress, UUID> versions = new ConcurrentHashMap<InetAddress, UUID>();
|
||||
final Set<InetAddress> liveHosts = Gossiper.instance.getLiveMembers();
|
||||
final Set<InetAddress> liveHosts = Gossiper.instance.getLiveEndpoints();
|
||||
final CountDownLatch latch = new CountDownLatch(liveHosts.size());
|
||||
|
||||
IAsyncCallback<UUID> cb = new IAsyncCallback<UUID>()
|
||||
|
|
@ -1963,7 +1963,7 @@ public class StorageProxy implements StorageProxyMBean
|
|||
|
||||
// maps versions to hosts that are on that version.
|
||||
Map<String, List<String>> results = new HashMap<String, List<String>>();
|
||||
Iterable<InetAddress> allHosts = Iterables.concat(Gossiper.instance.getLiveMembers(), Gossiper.instance.getUnreachableMembers());
|
||||
Iterable<InetAddress> allHosts = Iterables.concat(Gossiper.instance.getLiveEndpoints(), Gossiper.instance.getUnreachableMembers());
|
||||
for (InetAddress host : allHosts)
|
||||
{
|
||||
UUID version = versions.get(host);
|
||||
|
|
@ -2116,12 +2116,12 @@ public class StorageProxy implements StorageProxyMBean
|
|||
// Since the truncate operation is so aggressive and is typically only
|
||||
// invoked by an admin, for simplicity we require that all nodes are up
|
||||
// to perform the operation.
|
||||
int liveMembers = Gossiper.instance.getLiveMembers().size();
|
||||
int liveMembers = Gossiper.instance.getLiveEndpoints().size();
|
||||
throw new UnavailableException(ConsistencyLevel.ALL, liveMembers + Gossiper.instance.getUnreachableMembers().size(), liveMembers);
|
||||
}
|
||||
|
||||
Set<InetAddress> allEndpoints = Gossiper.instance.getLiveTokenOwners();
|
||||
|
||||
Set<InetAddress> allEndpoints = StorageService.instance.getLiveMembers(true);
|
||||
|
||||
int blockFor = allEndpoints.size();
|
||||
final TruncateResponseHandler responseHandler = new TruncateResponseHandler(blockFor);
|
||||
|
||||
|
|
|
|||
|
|
@ -17,8 +17,6 @@
|
|||
*/
|
||||
package org.apache.cassandra.service;
|
||||
|
||||
import static java.nio.charset.StandardCharsets.ISO_8859_1;
|
||||
|
||||
import java.io.ByteArrayInputStream;
|
||||
import java.io.DataInputStream;
|
||||
import java.io.File;
|
||||
|
|
@ -2477,9 +2475,33 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
|
|||
|
||||
public List<String> getLiveNodes()
|
||||
{
|
||||
return stringify(Gossiper.instance.getLiveMembers());
|
||||
return stringify(Gossiper.instance.getLiveEndpoints());
|
||||
}
|
||||
|
||||
public Set<InetAddress> getLiveMembers()
|
||||
{
|
||||
return getLiveMembers(false);
|
||||
}
|
||||
|
||||
public Set<InetAddress> getLiveMembers(boolean excludeDeadStates)
|
||||
{
|
||||
Set<InetAddress> ret = new HashSet<>();
|
||||
for (InetAddress ep : Gossiper.instance.getLiveEndpoints())
|
||||
{
|
||||
if (excludeDeadStates)
|
||||
{
|
||||
EndpointState epState = Gossiper.instance.getEndpointStateForEndpoint(ep);
|
||||
if (epState == null || Gossiper.instance.isDeadState(epState))
|
||||
continue;
|
||||
}
|
||||
|
||||
if (tokenMetadata.isMember(ep))
|
||||
ret.add(ep);
|
||||
}
|
||||
return ret;
|
||||
}
|
||||
|
||||
|
||||
public List<String> getUnreachableNodes()
|
||||
{
|
||||
return stringify(Gossiper.instance.getUnreachableMembers());
|
||||
|
|
@ -3773,7 +3795,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
|
|||
if (endpoint.equals(myAddress))
|
||||
throw new UnsupportedOperationException("Cannot remove self");
|
||||
|
||||
if (Gossiper.instance.getLiveMembers().contains(endpoint))
|
||||
if (Gossiper.instance.getLiveEndpoints().contains(endpoint))
|
||||
throw new UnsupportedOperationException("Node " + endpoint + " is alive and owns this ID. Use decommission command to remove it from the ring");
|
||||
|
||||
// A leaving endpoint that is dead is already being removed.
|
||||
|
|
|
|||
|
|
@ -18,15 +18,14 @@
|
|||
package org.apache.cassandra.locator;
|
||||
|
||||
import java.net.InetAddress;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.nio.file.Paths;
|
||||
|
||||
import com.google.common.net.InetAddresses;
|
||||
import org.apache.cassandra.utils.FBUtilities;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.apache.cassandra.utils.FBUtilities;
|
||||
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
/**
|
||||
* Unit tests for {@link GossipingPropertyFileSnitch}.
|
||||
*/
|
||||
|
|
@ -37,35 +36,14 @@ public class GossipingPropertyFileSnitchTest
|
|||
final String expectedRack)
|
||||
{
|
||||
final InetAddress endpoint = InetAddresses.forString(endpointString);
|
||||
Assert.assertEquals(expectedDatacenter, snitch.getDatacenter(endpoint));
|
||||
Assert.assertEquals(expectedRack, snitch.getRack(endpoint));
|
||||
assertEquals(expectedDatacenter, snitch.getDatacenter(endpoint));
|
||||
assertEquals(expectedRack, snitch.getRack(endpoint));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAutoReloadConfig() throws Exception
|
||||
public void testLoadConfig() throws Exception
|
||||
{
|
||||
String confFile = FBUtilities.resourceToFile(SnitchProperties.RACKDC_PROPERTY_FILENAME);
|
||||
|
||||
final GossipingPropertyFileSnitch snitch = new GossipingPropertyFileSnitch(/*refreshPeriodInSeconds*/1);
|
||||
final GossipingPropertyFileSnitch snitch = new GossipingPropertyFileSnitch();
|
||||
checkEndpoint(snitch, FBUtilities.getBroadcastAddress().getHostAddress(), "DC1", "RAC1");
|
||||
|
||||
final Path effectiveFile = Paths.get(confFile);
|
||||
final Path backupFile = Paths.get(confFile + ".bak");
|
||||
final Path modifiedFile = Paths.get(confFile + ".mod");
|
||||
|
||||
try
|
||||
{
|
||||
Files.copy(effectiveFile, backupFile);
|
||||
Files.copy(modifiedFile, effectiveFile, java.nio.file.StandardCopyOption.REPLACE_EXISTING);
|
||||
|
||||
Thread.sleep(1500);
|
||||
|
||||
checkEndpoint(snitch, FBUtilities.getBroadcastAddress().getHostAddress(), "DC2", "RAC2");
|
||||
}
|
||||
finally
|
||||
{
|
||||
Files.copy(backupFile, effectiveFile, java.nio.file.StandardCopyOption.REPLACE_EXISTING);
|
||||
Files.delete(backupFile);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,334 @@
|
|||
/*
|
||||
* 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 java.io.IOException;
|
||||
import java.net.InetAddress;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.nio.file.Paths;
|
||||
import java.nio.file.StandardOpenOption;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.UUID;
|
||||
|
||||
import com.google.common.net.InetAddresses;
|
||||
|
||||
import org.apache.cassandra.dht.IPartitioner;
|
||||
import org.apache.cassandra.dht.RandomPartitioner;
|
||||
import org.apache.cassandra.dht.Token;
|
||||
import org.apache.cassandra.exceptions.ConfigurationException;
|
||||
import org.apache.cassandra.gms.ApplicationState;
|
||||
import org.apache.cassandra.gms.Gossiper;
|
||||
import org.apache.cassandra.gms.VersionedValue;
|
||||
import org.apache.cassandra.service.StorageService;
|
||||
import org.apache.cassandra.utils.FBUtilities;
|
||||
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
/**
|
||||
* Unit tests for {@link PropertyFileSnitch}.
|
||||
*/
|
||||
public class PropertyFileSnitchTest
|
||||
{
|
||||
private Path effectiveFile;
|
||||
private Path backupFile;
|
||||
|
||||
private VersionedValue.VersionedValueFactory valueFactory;
|
||||
private Map<InetAddress, Set<Token>> tokenMap;
|
||||
|
||||
@Before
|
||||
public void setup() throws ConfigurationException, IOException
|
||||
{
|
||||
String confFile = FBUtilities.resourceToFile(PropertyFileSnitch.SNITCH_PROPERTIES_FILENAME);
|
||||
effectiveFile = Paths.get(confFile);
|
||||
backupFile = Paths.get(confFile + ".bak");
|
||||
|
||||
restoreOrigConfigFile();
|
||||
|
||||
InetAddress[] hosts = {
|
||||
InetAddress.getByName("127.0.0.1"), // this exists in the config file
|
||||
InetAddress.getByName("127.0.0.2"), // this exists in the config file
|
||||
InetAddress.getByName("127.0.0.9"), // this does not exist in the config file
|
||||
};
|
||||
|
||||
IPartitioner partitioner = new RandomPartitioner();
|
||||
valueFactory = new VersionedValue.VersionedValueFactory(partitioner);
|
||||
tokenMap = new HashMap<>();
|
||||
|
||||
for (InetAddress host : hosts)
|
||||
{
|
||||
Set<Token> tokens = Collections.singleton(partitioner.getRandomToken());
|
||||
Gossiper.instance.initializeNodeUnsafe(host, UUID.randomUUID(), 1);
|
||||
Gossiper.instance.injectApplicationState(host, ApplicationState.TOKENS, valueFactory.tokens(tokens));
|
||||
|
||||
setNodeShutdown(host);
|
||||
tokenMap.put(host, tokens);
|
||||
}
|
||||
}
|
||||
|
||||
private void restoreOrigConfigFile() throws IOException
|
||||
{
|
||||
if (Files.exists(backupFile))
|
||||
{
|
||||
Files.copy(backupFile, effectiveFile, java.nio.file.StandardCopyOption.REPLACE_EXISTING);
|
||||
Files.delete(backupFile);
|
||||
}
|
||||
}
|
||||
|
||||
private void replaceConfigFile(Map<String, String> replacements) throws IOException
|
||||
{
|
||||
List<String> lines = Files.readAllLines(effectiveFile, StandardCharsets.UTF_8);
|
||||
List<String> newLines = new ArrayList<>(lines.size());
|
||||
Set<String> replaced = new HashSet<>();
|
||||
|
||||
for (String line : lines)
|
||||
{
|
||||
String[] info = line.split("=");
|
||||
if (info.length == 2 && replacements.containsKey(info[0]))
|
||||
{
|
||||
String replacement = replacements.get(info[0]);
|
||||
if (!replacement.isEmpty()) // empty means remove this line
|
||||
newLines.add(info[0] + '=' + replacement);
|
||||
|
||||
replaced.add(info[0]);
|
||||
}
|
||||
else
|
||||
{
|
||||
newLines.add(line);
|
||||
}
|
||||
}
|
||||
|
||||
// add any new lines that were not replaced
|
||||
for (Map.Entry<String, String> replacement : replacements.entrySet())
|
||||
{
|
||||
if (replaced.contains(replacement.getKey()))
|
||||
continue;
|
||||
|
||||
if (!replacement.getValue().isEmpty()) // empty means remove this line so do nothing here
|
||||
newLines.add(replacement.getKey() + '=' + replacement.getValue());
|
||||
}
|
||||
|
||||
Files.write(effectiveFile, newLines, StandardCharsets.UTF_8, StandardOpenOption.TRUNCATE_EXISTING);
|
||||
}
|
||||
|
||||
private void setNodeShutdown(InetAddress host)
|
||||
{
|
||||
StorageService.instance.getTokenMetadata().removeEndpoint(host);
|
||||
Gossiper.instance.injectApplicationState(host, ApplicationState.STATUS, valueFactory.shutdown(true));
|
||||
Gossiper.instance.markDead(host, Gossiper.instance.getEndpointStateForEndpoint(host));
|
||||
}
|
||||
|
||||
private void setNodeLive(InetAddress host)
|
||||
{
|
||||
Gossiper.instance.injectApplicationState(host, ApplicationState.STATUS, valueFactory.normal(tokenMap.get(host)));
|
||||
Gossiper.instance.realMarkAlive(host, Gossiper.instance.getEndpointStateForEndpoint(host));
|
||||
StorageService.instance.getTokenMetadata().updateNormalTokens(tokenMap.get(host), host);
|
||||
}
|
||||
|
||||
private static void checkEndpoint(final AbstractNetworkTopologySnitch snitch,
|
||||
final String endpointString, final String expectedDatacenter,
|
||||
final String expectedRack)
|
||||
{
|
||||
final InetAddress endpoint = InetAddresses.forString(endpointString);
|
||||
assertEquals(expectedDatacenter, snitch.getDatacenter(endpoint));
|
||||
assertEquals(expectedRack, snitch.getRack(endpoint));
|
||||
}
|
||||
|
||||
/**
|
||||
* Test that changing rack for a host in the configuration file is only effective if the host is not live.
|
||||
* The original configuration file contains: 127.0.0.1=DC1:RAC1
|
||||
*/
|
||||
@Test
|
||||
public void testChangeHostRack() throws Exception
|
||||
{
|
||||
final InetAddress host = InetAddress.getByName("127.0.0.1");
|
||||
final PropertyFileSnitch snitch = new PropertyFileSnitch(/*refreshPeriodInSeconds*/1);
|
||||
checkEndpoint(snitch, host.getHostAddress(), "DC1", "RAC1");
|
||||
|
||||
try
|
||||
{
|
||||
setNodeLive(host);
|
||||
|
||||
Files.copy(effectiveFile, backupFile);
|
||||
replaceConfigFile(Collections.singletonMap(host.getHostAddress(), "DC1:RAC2"));
|
||||
|
||||
Thread.sleep(1500);
|
||||
checkEndpoint(snitch, host.getHostAddress(), "DC1", "RAC1");
|
||||
|
||||
setNodeShutdown(host);
|
||||
replaceConfigFile(Collections.singletonMap(host.getHostAddress(), "DC1:RAC2"));
|
||||
|
||||
Thread.sleep(1500);
|
||||
checkEndpoint(snitch, host.getHostAddress(), "DC1", "RAC2");
|
||||
}
|
||||
finally
|
||||
{
|
||||
restoreOrigConfigFile();
|
||||
setNodeShutdown(host);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Test that changing dc for a host in the configuration file is only effective if the host is not live.
|
||||
* The original configuration file contains: 127.0.0.1=DC1:RAC1
|
||||
*/
|
||||
@Test
|
||||
public void testChangeHostDc() throws Exception
|
||||
{
|
||||
final InetAddress host = InetAddress.getByName("127.0.0.1");
|
||||
final PropertyFileSnitch snitch = new PropertyFileSnitch(/*refreshPeriodInSeconds*/1);
|
||||
checkEndpoint(snitch, host.getHostAddress(), "DC1", "RAC1");
|
||||
|
||||
try
|
||||
{
|
||||
setNodeLive(host);
|
||||
|
||||
Files.copy(effectiveFile, backupFile);
|
||||
replaceConfigFile(Collections.singletonMap(host.getHostAddress(), "DC2:RAC1"));
|
||||
|
||||
Thread.sleep(1500);
|
||||
checkEndpoint(snitch, host.getHostAddress(), "DC1", "RAC1");
|
||||
|
||||
setNodeShutdown(host);
|
||||
replaceConfigFile(Collections.singletonMap(host.getHostAddress(), "DC2:RAC1"));
|
||||
|
||||
Thread.sleep(1500);
|
||||
checkEndpoint(snitch, host.getHostAddress(), "DC2", "RAC1");
|
||||
}
|
||||
finally
|
||||
{
|
||||
restoreOrigConfigFile();
|
||||
setNodeShutdown(host);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Test that adding a host to the configuration file changes the host dc and rack only if the host
|
||||
* is not live. The original configuration file does not contain 127.0.0.9 and so it should use
|
||||
* the default default=DC1:r1.
|
||||
*/
|
||||
@Test
|
||||
public void testAddHost() throws Exception
|
||||
{
|
||||
final InetAddress host = InetAddress.getByName("127.0.0.9");
|
||||
final PropertyFileSnitch snitch = new PropertyFileSnitch(/*refreshPeriodInSeconds*/1);
|
||||
checkEndpoint(snitch, host.getHostAddress(), "DC1", "r1"); // default
|
||||
|
||||
try
|
||||
{
|
||||
setNodeLive(host);
|
||||
|
||||
Files.copy(effectiveFile, backupFile);
|
||||
replaceConfigFile(Collections.singletonMap(host.getHostAddress(), "DC2:RAC2")); // add this line if not yet there
|
||||
|
||||
Thread.sleep(1500);
|
||||
checkEndpoint(snitch, host.getHostAddress(), "DC1", "r1"); // unchanged
|
||||
|
||||
setNodeShutdown(host);
|
||||
replaceConfigFile(Collections.singletonMap(host.getHostAddress(), "DC2:RAC2")); // add this line if not yet there
|
||||
|
||||
Thread.sleep(1500);
|
||||
checkEndpoint(snitch, host.getHostAddress(), "DC2", "RAC2"); // changed
|
||||
}
|
||||
finally
|
||||
{
|
||||
restoreOrigConfigFile();
|
||||
setNodeShutdown(host);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Test that removing a host from the configuration file changes the host rack only if the host
|
||||
* is not live. The original configuration file contains 127.0.0.2=DC1:RAC2 and default=DC1:r1 so removing
|
||||
* this host should result in a different rack if the host is not live.
|
||||
*/
|
||||
@Test
|
||||
public void testRemoveHost() throws Exception
|
||||
{
|
||||
final InetAddress host = InetAddress.getByName("127.0.0.2");
|
||||
final PropertyFileSnitch snitch = new PropertyFileSnitch(/*refreshPeriodInSeconds*/1);
|
||||
checkEndpoint(snitch, host.getHostAddress(), "DC1", "RAC2");
|
||||
|
||||
try
|
||||
{
|
||||
setNodeLive(host);
|
||||
|
||||
Files.copy(effectiveFile, backupFile);
|
||||
replaceConfigFile(Collections.singletonMap(host.getHostAddress(), "")); // removes line if found
|
||||
|
||||
Thread.sleep(1500);
|
||||
checkEndpoint(snitch, host.getHostAddress(), "DC1", "RAC2"); // unchanged
|
||||
|
||||
setNodeShutdown(host);
|
||||
replaceConfigFile(Collections.singletonMap(host.getHostAddress(), "")); // removes line if found
|
||||
|
||||
Thread.sleep(1500);
|
||||
checkEndpoint(snitch, host.getHostAddress(), "DC1", "r1"); // default
|
||||
}
|
||||
finally
|
||||
{
|
||||
restoreOrigConfigFile();
|
||||
setNodeShutdown(host);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Test that we can change the default only if this does not result in any live node changing dc or rack.
|
||||
* The configuration file contains default=DC1:r1 and we change it to default=DC2:r2. Let's use host 127.0.0.9
|
||||
* since it is not in the configuration file.
|
||||
*/
|
||||
@Test
|
||||
public void testChangeDefault() throws Exception
|
||||
{
|
||||
final InetAddress host = InetAddress.getByName("127.0.0.9");
|
||||
final PropertyFileSnitch snitch = new PropertyFileSnitch(/*refreshPeriodInSeconds*/1);
|
||||
checkEndpoint(snitch, host.getHostAddress(), "DC1", "r1"); // default
|
||||
|
||||
try
|
||||
{
|
||||
setNodeLive(host);
|
||||
|
||||
Files.copy(effectiveFile, backupFile);
|
||||
replaceConfigFile(Collections.singletonMap("default", "DC2:r2")); // change default
|
||||
|
||||
Thread.sleep(1500);
|
||||
checkEndpoint(snitch, host.getHostAddress(), "DC1", "r1"); // unchanged
|
||||
|
||||
setNodeShutdown(host);
|
||||
replaceConfigFile(Collections.singletonMap("default", "DC2:r2")); // change default again (refresh file update)
|
||||
|
||||
Thread.sleep(1500);
|
||||
checkEndpoint(snitch, host.getHostAddress(), "DC2", "r2"); // default updated
|
||||
}
|
||||
finally
|
||||
{
|
||||
restoreOrigConfigFile();
|
||||
setNodeShutdown(host);
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue