diff --git a/CHANGES.txt b/CHANGES.txt
index b29ae784b2..d8ae88ff19 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
4.0
+ * Allow storage port to be configurable per node (CASSANDRA-7544)
* Make sub-range selection for non-frozen collections return null instead of empty (CASSANDRA-14182)
* BloomFilter serialization format should not change byte ordering (CASSANDRA-9067)
* Remove unused on-heap BloomFilter implementation (CASSANDRA-14152)
diff --git a/NEWS.txt b/NEWS.txt
index 06d73eace2..c314030656 100644
--- a/NEWS.txt
+++ b/NEWS.txt
@@ -39,7 +39,7 @@ New features
This property can be modified at runtime through both JMX and the new `setconcurrentviewbuilders`
and `getconcurrentviewbuilders` nodetool commands. See CASSANDRA-12245 for more details.
- There is now a binary full query log based on Chronicle Queue that can be controlled using
- nodetool enablefullquerylog, disablefullquerylog, and resetfullquerylog. The log
+ nodetool enablefullquerylog, disablefullquerylog, and resetfullquerylog. The log
contains all queries invoked, approximate time they were invoked, any parameters necessary
to bind wildcard values, and all query options. A human readable version of the log can be
dumped or tailed using the new bin/fqltool utility. The full query log is designed to be safe
@@ -103,6 +103,22 @@ Upgrading
but blocks for up to a configurable number of milliseconds between disk flushes.
- nodetool clearsnapshot now required the --all flag to remove all snapshots.
Previous behavior would delete all snapshots by default.
+ - Nodes are now identified by a combination of IP, and storage port.
+ Existing JMX APIs, nodetool, and system tables continue to work
+ and accept/return just an IP, but there is a new
+ version of each that works with the full unambiguous identifier.
+ You should prefer these over the deprecated ambiguous versions that only
+ work with an IP. This was done to support multiple instances per IP.
+ Additionally we are moving to only using a single port for encrypted and
+ unencrypted traffic and if you want multiple instances per IP you must
+ first switch encrypted traffic to the storage port and not a separate
+ encrypted port. If you want to use multiple instances per IP
+ with SSL you will need to use StartTLS on storage_port and set
+ outgoing_encrypted_port_source to gossip outbound connections
+ know what port to connect to for each instance. Before changing
+ storage port or native port at nodes you must first upgrade the entire cluster
+ and clients to 4.0 so they can handle the port not being consistent across
+ the cluster.
Materialized Views
-------------------
diff --git a/bin/cqlsh.py b/bin/cqlsh.py
index 61ea1605ce..3055110d1c 100644
--- a/bin/cqlsh.py
+++ b/bin/cqlsh.py
@@ -455,7 +455,8 @@ class Shell(cmd.Cmd):
single_statement=None,
request_timeout=DEFAULT_REQUEST_TIMEOUT_SECONDS,
protocol_version=None,
- connect_timeout=DEFAULT_CONNECT_TIMEOUT_SECONDS):
+ connect_timeout=DEFAULT_CONNECT_TIMEOUT_SECONDS,
+ allow_server_port_discovery=False):
cmd.Cmd.__init__(self, completekey=completekey)
self.hostname = hostname
self.port = port
@@ -470,6 +471,7 @@ class Shell(cmd.Cmd):
self.tracing_enabled = tracing_enabled
self.page_size = self.default_page_size
self.expand_enabled = expand_enabled
+ self.allow_server_port_discovery = allow_server_port_discovery
if use_conn:
self.conn = use_conn
else:
@@ -482,6 +484,7 @@ class Shell(cmd.Cmd):
load_balancing_policy=WhiteListRoundRobinPolicy([self.hostname]),
control_connection_timeout=connect_timeout,
connect_timeout=connect_timeout,
+ allow_server_port_discovery=allow_server_port_discovery,
**kwargs)
self.owns_connection = not use_conn
@@ -1768,7 +1771,8 @@ class Shell(cmd.Cmd):
display_timezone=self.display_timezone,
max_trace_wait=self.max_trace_wait, ssl=self.ssl,
request_timeout=self.session.default_timeout,
- connect_timeout=self.conn.connect_timeout)
+ connect_timeout=self.conn.connect_timeout,
+ allow_server_port_discovery=self.allow_server_port_discovery)
subshell.cmdloop()
f.close()
@@ -2252,6 +2256,7 @@ def read_options(cmdlineargs, environment):
optvalues.connect_timeout = option_with_default(configs.getint, 'connection', 'timeout', DEFAULT_CONNECT_TIMEOUT_SECONDS)
optvalues.request_timeout = option_with_default(configs.getint, 'connection', 'request_timeout', DEFAULT_REQUEST_TIMEOUT_SECONDS)
optvalues.execute = None
+ optvalues.allow_server_port_discovery = option_with_default(configs.getboolean, 'connection', 'allow_server_port_discovery', 'False')
(options, arguments) = parser.parse_args(cmdlineargs, values=optvalues)
@@ -2415,7 +2420,8 @@ def main(options, hostname, port):
single_statement=options.execute,
request_timeout=options.request_timeout,
connect_timeout=options.connect_timeout,
- encoding=options.encoding)
+ encoding=options.encoding,
+ allow_server_port_discovery=options.allow_server_port_discovery)
except KeyboardInterrupt:
sys.exit('Connection aborted.')
except CQL_ERRORS, e:
diff --git a/build.xml b/build.xml
index 8a033e0e54..579686888a 100644
--- a/build.xml
+++ b/build.xml
@@ -306,7 +306,7 @@
-
+
@@ -437,7 +437,7 @@
@@ -543,7 +543,7 @@
+ -->
@@ -624,7 +624,7 @@
-->
-
+
diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml
index 80e4515431..3bed3a6bd1 100644
--- a/conf/cassandra.yaml
+++ b/conf/cassandra.yaml
@@ -416,7 +416,7 @@ seed_provider:
parameters:
# seeds is actually a comma-delimited list of addresses.
# Ex: ",,"
- - seeds: "127.0.0.1"
+ - seeds: "127.0.0.1:7000"
# For workloads with more data than can fit in memory, Cassandra's
# bottleneck will be reads that need to fetch data from
@@ -945,6 +945,13 @@ dynamic_snitch_badness_threshold: 0.1
# the keystore and truststore. For instructions on generating these files, see:
# http://download.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html#CreateKeystore
#
+# If you are taking advantage of StartTLS outbound connections will have the issue that they can't know
+# what encrypted port to connect to in a foolproof way. outgoing_encrypted_port_source deals with this confusion
+# by allowing you to specify how you want a node to pick an outgoing port for intra-cluster connections.
+# Valid values are "gossip" and "yaml". Gossip will always connect to the storage port for a node that is
+# published via a gossip which is always going to be the plain storage port. "yaml" will always select
+# the port configured as ssl_storage_port on THIS node. If you want to use SSL and have different storage
+# ports across the cluster you must select "gossip" and use StartTLS on storage_port.
server_encryption_options:
# set to true for allowing secure incoming connections
enabled: false
diff --git a/lib/cassandra-driver-core-3.3.2-0461ed35-SNAPSHOT-shaded.jar b/lib/cassandra-driver-core-4.0.0-SNAPSHOT-shaded.jar
similarity index 78%
rename from lib/cassandra-driver-core-3.3.2-0461ed35-SNAPSHOT-shaded.jar
rename to lib/cassandra-driver-core-4.0.0-SNAPSHOT-shaded.jar
index a7be9cbe9e..609c393f35 100644
Binary files a/lib/cassandra-driver-core-3.3.2-0461ed35-SNAPSHOT-shaded.jar and b/lib/cassandra-driver-core-4.0.0-SNAPSHOT-shaded.jar differ
diff --git a/lib/cassandra-driver-internal-only-3.11.zip b/lib/cassandra-driver-internal-only-3.12.0.post0-9ee88ded.zip
similarity index 50%
rename from lib/cassandra-driver-internal-only-3.11.zip
rename to lib/cassandra-driver-internal-only-3.12.0.post0-9ee88ded.zip
index f7760af405..4aa91b7046 100644
Binary files a/lib/cassandra-driver-internal-only-3.11.zip and b/lib/cassandra-driver-internal-only-3.12.0.post0-9ee88ded.zip differ
diff --git a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
index 807e970fa3..f232bdc653 100644
--- a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
+++ b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
@@ -19,7 +19,6 @@ package org.apache.cassandra.batchlog;
import java.io.IOException;
import java.lang.management.ManagementFactory;
-import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.*;
@@ -34,6 +33,7 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.DebuggableScheduledThreadPoolExecutor;
import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.db.*;
@@ -250,7 +250,7 @@ public class BatchlogManager implements BatchlogManagerMBean
int positionInPage = 0;
ArrayList unfinishedBatches = new ArrayList<>(pageSize);
- Set hintedNodes = new HashSet<>();
+ Set hintedNodes = new HashSet<>();
Set replayedBatches = new HashSet<>();
// Sending out batches for replay without waiting for them, so that one stuck batch doesn't affect others
@@ -295,7 +295,7 @@ public class BatchlogManager implements BatchlogManagerMBean
replayedBatches.forEach(BatchlogManager::remove);
}
- private void finishAndClearBatches(ArrayList batches, Set hintedNodes, Set replayedBatches)
+ private void finishAndClearBatches(ArrayList batches, Set hintedNodes, Set replayedBatches)
{
// schedule hints for timed out deliveries
for (ReplayingBatch batch : batches)
@@ -330,7 +330,7 @@ public class BatchlogManager implements BatchlogManagerMBean
this.replayedBytes = addMutations(version, serializedMutations);
}
- public int replay(RateLimiter rateLimiter, Set hintedNodes) throws IOException
+ public int replay(RateLimiter rateLimiter, Set hintedNodes) throws IOException
{
logger.trace("Replaying batch {}", id);
@@ -348,7 +348,7 @@ public class BatchlogManager implements BatchlogManagerMBean
return replayHandlers.size();
}
- public void finish(Set hintedNodes)
+ public void finish(Set hintedNodes)
{
for (int i = 0; i < replayHandlers.size(); i++)
{
@@ -396,7 +396,7 @@ public class BatchlogManager implements BatchlogManagerMBean
mutations.add(mutation);
}
- private void writeHintsForUndeliveredEndpoints(int startFrom, Set hintedNodes)
+ private void writeHintsForUndeliveredEndpoints(int startFrom, Set hintedNodes)
{
int gcgs = gcgs(mutations);
@@ -420,7 +420,7 @@ public class BatchlogManager implements BatchlogManagerMBean
private static List> sendReplays(List mutations,
long writtenAt,
- Set hintedNodes)
+ Set hintedNodes)
{
List> handlers = new ArrayList<>(mutations.size());
for (Mutation mutation : mutations)
@@ -440,15 +440,15 @@ public class BatchlogManager implements BatchlogManagerMBean
*/
private static ReplayWriteResponseHandler sendSingleReplayMutation(final Mutation mutation,
long writtenAt,
- Set hintedNodes)
+ Set hintedNodes)
{
- Set liveEndpoints = new HashSet<>();
+ Set liveEndpoints = new HashSet<>();
String ks = mutation.getKeyspaceName();
Token tk = mutation.key().getToken();
- for (InetAddress endpoint : StorageService.instance.getNaturalAndPendingEndpoints(ks, tk))
+ for (InetAddressAndPort endpoint : StorageService.instance.getNaturalAndPendingEndpoints(ks, tk))
{
- if (endpoint.equals(FBUtilities.getBroadcastAddress()))
+ if (endpoint.equals(FBUtilities.getBroadcastAddressAndPort()))
{
mutation.apply();
}
@@ -469,7 +469,7 @@ public class BatchlogManager implements BatchlogManagerMBean
ReplayWriteResponseHandler handler = new ReplayWriteResponseHandler<>(liveEndpoints, System.nanoTime());
MessageOut message = mutation.createMessage();
- for (InetAddress endpoint : liveEndpoints)
+ for (InetAddressAndPort endpoint : liveEndpoints)
MessagingService.instance().sendRR(message, endpoint, handler, false);
return handler;
}
@@ -488,11 +488,11 @@ public class BatchlogManager implements BatchlogManagerMBean
*/
private static class ReplayWriteResponseHandler extends WriteResponseHandler
{
- private final Set undelivered = Collections.newSetFromMap(new ConcurrentHashMap<>());
+ private final Set undelivered = Collections.newSetFromMap(new ConcurrentHashMap<>());
- ReplayWriteResponseHandler(Collection writeEndpoints, long queryStartNanoTime)
+ ReplayWriteResponseHandler(Collection writeEndpoints, long queryStartNanoTime)
{
- super(writeEndpoints, Collections.emptySet(), null, null, null, WriteType.UNLOGGED_BATCH, queryStartNanoTime);
+ super(writeEndpoints, Collections.emptySet(), null, null, null, WriteType.UNLOGGED_BATCH, queryStartNanoTime);
undelivered.addAll(writeEndpoints);
}
@@ -505,7 +505,7 @@ public class BatchlogManager implements BatchlogManagerMBean
@Override
public void response(MessageIn m)
{
- boolean removed = undelivered.remove(m == null ? FBUtilities.getBroadcastAddress() : m.from);
+ boolean removed = undelivered.remove(m == null ? FBUtilities.getBroadcastAddressAndPort() : m.from);
assert removed;
super.response(m);
}
@@ -515,9 +515,9 @@ public class BatchlogManager implements BatchlogManagerMBean
public static class EndpointFilter
{
private final String localRack;
- private final Multimap endpoints;
+ private final Multimap endpoints;
- public EndpointFilter(String localRack, Multimap endpoints)
+ public EndpointFilter(String localRack, Multimap endpoints)
{
this.localRack = localRack;
this.endpoints = endpoints;
@@ -526,15 +526,15 @@ public class BatchlogManager implements BatchlogManagerMBean
/**
* @return list of candidates for batchlog hosting. If possible these will be two nodes from different racks.
*/
- public Collection filter()
+ public Collection filter()
{
// special case for single-node data centers
if (endpoints.values().size() == 1)
return endpoints.values();
// strip out dead endpoints and localhost
- ListMultimap validated = ArrayListMultimap.create();
- for (Map.Entry entry : endpoints.entries())
+ ListMultimap validated = ArrayListMultimap.create();
+ for (Map.Entry entry : endpoints.entries())
if (isValid(entry.getValue()))
validated.put(entry.getKey(), entry.getValue());
@@ -554,7 +554,7 @@ public class BatchlogManager implements BatchlogManagerMBean
* pick two random nodes from there; we are guaranteed to have at least two nodes in the single remaining rack
* because of the preceding if block.
*/
- List otherRack = Lists.newArrayList(validated.values());
+ List otherRack = Lists.newArrayList(validated.values());
shuffle(otherRack);
return otherRack.subList(0, 2);
}
@@ -572,10 +572,10 @@ public class BatchlogManager implements BatchlogManagerMBean
}
// grab a random member of up to two racks
- List result = new ArrayList<>(2);
+ List result = new ArrayList<>(2);
for (String rack : Iterables.limit(racks, 2))
{
- List rackMembers = validated.get(rack);
+ List rackMembers = validated.get(rack);
result.add(rackMembers.get(getRandomInt(rackMembers.size())));
}
@@ -583,9 +583,9 @@ public class BatchlogManager implements BatchlogManagerMBean
}
@VisibleForTesting
- protected boolean isValid(InetAddress input)
+ protected boolean isValid(InetAddressAndPort input)
{
- return !input.equals(FBUtilities.getBroadcastAddress()) && FailureDetector.instance.isAlive(input);
+ return !input.equals(FBUtilities.getBroadcastAddressAndPort()) && FailureDetector.instance.isAlive(input);
}
@VisibleForTesting
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index a656d1f1ba..9012e3ac88 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -55,6 +55,7 @@ import org.apache.cassandra.io.util.SsdDiskOptimizationStrategy;
import org.apache.cassandra.locator.DynamicEndpointSnitch;
import org.apache.cassandra.locator.EndpointSnitchInfo;
import org.apache.cassandra.locator.IEndpointSnitch;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.SeedProvider;
import org.apache.cassandra.net.BackPressureStrategy;
import org.apache.cassandra.net.RateBasedBackPressure;
@@ -110,7 +111,7 @@ public class DatabaseDescriptor
private static long indexSummaryCapacityInMB;
private static String localDC;
- private static Comparator localComparator;
+ private static Comparator localComparator;
private static EncryptionContext encryptionContext;
private static boolean hasLoggedConfig;
@@ -307,6 +308,7 @@ public class DatabaseDescriptor
private static void applyAll() throws ConfigurationException
{
+ //InetAddressAndPort cares that applySimpleConfig runs first
applySimpleConfig();
applyPartitioner();
@@ -324,6 +326,9 @@ public class DatabaseDescriptor
private static void applySimpleConfig()
{
+ //Doing this first before all other things in case other pieces of config want to construct
+ //InetAddressAndPort and get the right defaults
+ InetAddressAndPort.initializeDefaultPort(getStoragePort());
if (conf.commitlog_sync == null)
{
@@ -827,7 +832,7 @@ public class DatabaseDescriptor
}
else
{
- rpcAddress = FBUtilities.getLocalAddress();
+ rpcAddress = FBUtilities.getJustLocalAddress();
}
/* RPC address to broadcast */
@@ -956,10 +961,10 @@ public class DatabaseDescriptor
snitch = createEndpointSnitch(conf.dynamic_snitch, conf.endpoint_snitch);
EndpointSnitchInfo.create();
- localDC = snitch.getDatacenter(FBUtilities.getBroadcastAddress());
- localComparator = new Comparator()
+ localDC = snitch.getDatacenter(FBUtilities.getBroadcastAddressAndPort());
+ localComparator = new Comparator()
{
- public int compare(InetAddress endpoint1, InetAddress endpoint2)
+ public int compare(InetAddressAndPort endpoint1, InetAddressAndPort endpoint2)
{
boolean local1 = localDC.equals(snitch.getDatacenter(endpoint1));
boolean local2 = localDC.equals(snitch.getDatacenter(endpoint2));
@@ -1319,14 +1324,14 @@ public class DatabaseDescriptor
return conf.num_tokens;
}
- public static InetAddress getReplaceAddress()
+ public static InetAddressAndPort getReplaceAddress()
{
try
{
if (System.getProperty(Config.PROPERTY_PREFIX + "replace_address", null) != null)
- return InetAddress.getByName(System.getProperty(Config.PROPERTY_PREFIX + "replace_address", null));
+ return InetAddressAndPort.getByName(System.getProperty(Config.PROPERTY_PREFIX + "replace_address", null));
else if (System.getProperty(Config.PROPERTY_PREFIX + "replace_address_first_boot", null) != null)
- return InetAddress.getByName(System.getProperty(Config.PROPERTY_PREFIX + "replace_address_first_boot", null));
+ return InetAddressAndPort.getByName(System.getProperty(Config.PROPERTY_PREFIX + "replace_address_first_boot", null));
return null;
}
catch (UnknownHostException e)
@@ -1651,9 +1656,9 @@ public class DatabaseDescriptor
return conf.saved_caches_directory;
}
- public static Set getSeeds()
+ public static Set getSeeds()
{
- return ImmutableSet.builder().addAll(seedProvider.getSeeds()).build();
+ return ImmutableSet.builder().addAll(seedProvider.getSeeds()).build();
}
public static InetAddress getListenAddress()
@@ -2206,7 +2211,7 @@ public class DatabaseDescriptor
return localDC;
}
- public static Comparator getLocalComparator()
+ public static Comparator getLocalComparator()
{
return localComparator;
}
diff --git a/src/java/org/apache/cassandra/db/ConsistencyLevel.java b/src/java/org/apache/cassandra/db/ConsistencyLevel.java
index 2214d8d64a..f93e737ebd 100644
--- a/src/java/org/apache/cassandra/db/ConsistencyLevel.java
+++ b/src/java/org/apache/cassandra/db/ConsistencyLevel.java
@@ -17,7 +17,6 @@
*/
package org.apache.cassandra.db;
-import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -28,6 +27,7 @@ import com.google.common.collect.Iterables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.service.ReadRepairDecision;
@@ -145,21 +145,21 @@ public enum ConsistencyLevel
return isDCLocal;
}
- public boolean isLocal(InetAddress endpoint)
+ public boolean isLocal(InetAddressAndPort endpoint)
{
return DatabaseDescriptor.getLocalDataCenter().equals(DatabaseDescriptor.getEndpointSnitch().getDatacenter(endpoint));
}
- public int countLocalEndpoints(Iterable liveEndpoints)
+ public int countLocalEndpoints(Iterable liveEndpoints)
{
int count = 0;
- for (InetAddress endpoint : liveEndpoints)
+ for (InetAddressAndPort endpoint : liveEndpoints)
if (isLocal(endpoint))
count++;
return count;
}
- private Map countPerDCEndpoints(Keyspace keyspace, Iterable liveEndpoints)
+ private Map countPerDCEndpoints(Keyspace keyspace, Iterable liveEndpoints)
{
NetworkTopologyStrategy strategy = (NetworkTopologyStrategy) keyspace.getReplicationStrategy();
@@ -167,7 +167,7 @@ public enum ConsistencyLevel
for (String dc: strategy.getDatacenters())
dcEndpoints.put(dc, 0);
- for (InetAddress endpoint : liveEndpoints)
+ for (InetAddressAndPort endpoint : liveEndpoints)
{
String dc = DatabaseDescriptor.getEndpointSnitch().getDatacenter(endpoint);
dcEndpoints.put(dc, dcEndpoints.get(dc) + 1);
@@ -175,12 +175,12 @@ public enum ConsistencyLevel
return dcEndpoints;
}
- public List filterForQuery(Keyspace keyspace, List liveEndpoints)
+ public List filterForQuery(Keyspace keyspace, List liveEndpoints)
{
return filterForQuery(keyspace, liveEndpoints, ReadRepairDecision.NONE);
}
- public List filterForQuery(Keyspace keyspace, List liveEndpoints, ReadRepairDecision readRepair)
+ public List filterForQuery(Keyspace keyspace, List liveEndpoints, ReadRepairDecision readRepair)
{
/*
* If we are doing an each quorum query, we have to make sure that the endpoints we select
@@ -206,9 +206,9 @@ public enum ConsistencyLevel
case GLOBAL:
return liveEndpoints;
case DC_LOCAL:
- List local = new ArrayList();
- List other = new ArrayList();
- for (InetAddress add : liveEndpoints)
+ List local = new ArrayList<>();
+ List other = new ArrayList<>();
+ for (InetAddressAndPort add : liveEndpoints)
{
if (isLocal(add))
local.add(add);
@@ -225,7 +225,7 @@ public enum ConsistencyLevel
}
}
- private List filterForEachQuorum(Keyspace keyspace, List liveEndpoints, ReadRepairDecision readRepair)
+ private List filterForEachQuorum(Keyspace keyspace, List liveEndpoints, ReadRepairDecision readRepair)
{
NetworkTopologyStrategy strategy = (NetworkTopologyStrategy) keyspace.getReplicationStrategy();
@@ -233,20 +233,20 @@ public enum ConsistencyLevel
if (readRepair == ReadRepairDecision.GLOBAL)
return liveEndpoints;
- Map> dcsEndpoints = new HashMap<>();
+ Map> dcsEndpoints = new HashMap<>();
for (String dc: strategy.getDatacenters())
dcsEndpoints.put(dc, new ArrayList<>());
- for (InetAddress add : liveEndpoints)
+ for (InetAddressAndPort add : liveEndpoints)
{
String dc = DatabaseDescriptor.getEndpointSnitch().getDatacenter(add);
dcsEndpoints.get(dc).add(add);
}
- List waitSet = new ArrayList<>();
- for (Map.Entry> dcEndpoints : dcsEndpoints.entrySet())
+ List waitSet = new ArrayList<>();
+ for (Map.Entry> dcEndpoints : dcsEndpoints.entrySet())
{
- List dcEndpoint = dcEndpoints.getValue();
+ List dcEndpoint = dcEndpoints.getValue();
if (readRepair == ReadRepairDecision.DC_LOCAL && dcEndpoints.getKey().equals(DatabaseDescriptor.getLocalDataCenter()))
waitSet.addAll(dcEndpoint);
else
@@ -256,7 +256,7 @@ public enum ConsistencyLevel
return waitSet;
}
- public boolean isSufficientLiveNodes(Keyspace keyspace, Iterable liveEndpoints)
+ public boolean isSufficientLiveNodes(Keyspace keyspace, Iterable liveEndpoints)
{
switch (this)
{
@@ -283,7 +283,7 @@ public enum ConsistencyLevel
}
}
- public void assureSufficientLiveNodes(Keyspace keyspace, Iterable liveEndpoints) throws UnavailableException
+ public void assureSufficientLiveNodes(Keyspace keyspace, Iterable liveEndpoints) throws UnavailableException
{
int blockFor = blockFor(keyspace);
switch (this)
@@ -302,7 +302,7 @@ public enum ConsistencyLevel
if (logger.isTraceEnabled())
{
StringBuilder builder = new StringBuilder("Local replicas [");
- for (InetAddress endpoint : liveEndpoints)
+ for (InetAddressAndPort endpoint : liveEndpoints)
{
if (isLocal(endpoint))
builder.append(endpoint).append(",");
diff --git a/src/java/org/apache/cassandra/db/CounterMutationVerbHandler.java b/src/java/org/apache/cassandra/db/CounterMutationVerbHandler.java
index bd273e4255..95d791618a 100644
--- a/src/java/org/apache/cassandra/db/CounterMutationVerbHandler.java
+++ b/src/java/org/apache/cassandra/db/CounterMutationVerbHandler.java
@@ -37,7 +37,7 @@ public class CounterMutationVerbHandler implements IVerbHandler
final CounterMutation cm = message.payload;
logger.trace("Applying forwarded {}", cm);
- String localDataCenter = DatabaseDescriptor.getEndpointSnitch().getDatacenter(FBUtilities.getBroadcastAddress());
+ String localDataCenter = DatabaseDescriptor.getEndpointSnitch().getDatacenter(FBUtilities.getBroadcastAddressAndPort());
// We should not wait for the result of the write in this thread,
// otherwise we could have a distributed deadlock between replicas
// running this VerbHandler (see #4578).
diff --git a/src/java/org/apache/cassandra/db/DiskBoundaryManager.java b/src/java/org/apache/cassandra/db/DiskBoundaryManager.java
index 03cbf7bb90..72b5e2ae89 100644
--- a/src/java/org/apache/cassandra/db/DiskBoundaryManager.java
+++ b/src/java/org/apache/cassandra/db/DiskBoundaryManager.java
@@ -80,14 +80,14 @@ public class DiskBoundaryManager
&& !StorageService.isReplacingSameAddress()) // When replacing same address, the node marks itself as UN locally
{
PendingRangeCalculatorService.instance.blockUntilFinished();
- localRanges = tmd.getPendingRanges(cfs.keyspace.getName(), FBUtilities.getBroadcastAddress());
+ localRanges = tmd.getPendingRanges(cfs.keyspace.getName(), FBUtilities.getBroadcastAddressAndPort());
}
else
{
// Reason we use use the future settled TMD is that if we decommission a node, we want to stream
// from that node to the correct location on disk, if we didn't, we would put new files in the wrong places.
// We do this to minimize the amount of data we need to move in rebalancedisks once everything settled
- localRanges = cfs.keyspace.getReplicationStrategy().getAddressRanges(tmd.cloneAfterAllSettled()).get(FBUtilities.getBroadcastAddress());
+ localRanges = cfs.keyspace.getReplicationStrategy().getAddressRanges(tmd.cloneAfterAllSettled()).get(FBUtilities.getBroadcastAddressAndPort());
}
logger.debug("Got local ranges {} (ringVersion = {})", localRanges, ringVersion);
}
diff --git a/src/java/org/apache/cassandra/db/Keyspace.java b/src/java/org/apache/cassandra/db/Keyspace.java
index cebf6ebce6..ae778f1b5c 100644
--- a/src/java/org/apache/cassandra/db/Keyspace.java
+++ b/src/java/org/apache/cassandra/db/Keyspace.java
@@ -352,6 +352,7 @@ public class Keyspace
private void createReplicationStrategy(KeyspaceMetadata ksm)
{
+ logger.info("Creating replication strategy " + ksm .name + " params " + ksm.params);
replicationStrategy = AbstractReplicationStrategy.createReplicationStrategy(ksm.name,
ksm.params.replication.klass,
StorageService.instance.getTokenMetadata(),
diff --git a/src/java/org/apache/cassandra/db/Mutation.java b/src/java/org/apache/cassandra/db/Mutation.java
index a6a920c213..6195fe4c87 100644
--- a/src/java/org/apache/cassandra/db/Mutation.java
+++ b/src/java/org/apache/cassandra/db/Mutation.java
@@ -43,9 +43,6 @@ public class Mutation implements IMutation
{
public static final MutationSerializer serializer = new MutationSerializer();
- public static final String FORWARD_TO = "FWD_TO";
- public static final String FORWARD_FROM = "FWD_FRM";
-
// todo this is redundant
// when we remove it, also restore SerializationsTest.testMutationRead to not regenerate new Mutations each test
private final String keyspaceName;
diff --git a/src/java/org/apache/cassandra/db/MutationVerbHandler.java b/src/java/org/apache/cassandra/db/MutationVerbHandler.java
index 59247a25d2..8386048f60 100644
--- a/src/java/org/apache/cassandra/db/MutationVerbHandler.java
+++ b/src/java/org/apache/cassandra/db/MutationVerbHandler.java
@@ -17,18 +17,17 @@
*/
package org.apache.cassandra.db;
-import java.io.DataInputStream;
import java.io.IOException;
-import java.net.InetAddress;
+import java.util.Iterator;
import org.apache.cassandra.exceptions.WriteTimeoutException;
-import org.apache.cassandra.io.util.FastByteArrayInputStream;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.*;
import org.apache.cassandra.tracing.Tracing;
public class MutationVerbHandler implements IVerbHandler
{
- private void reply(int id, InetAddress replyTo)
+ private void reply(int id, InetAddressAndPort replyTo)
{
Tracing.trace("Enqueuing response to {}", replyTo);
MessagingService.instance().sendReply(WriteResponse.createMessage(), id, replyTo);
@@ -42,18 +41,19 @@ public class MutationVerbHandler implements IVerbHandler
public void doVerb(MessageIn message, int id) throws IOException
{
// Check if there were any forwarding headers in this message
- byte[] from = message.parameters.get(Mutation.FORWARD_FROM);
- InetAddress replyTo;
+ InetAddressAndPort from = (InetAddressAndPort)message.parameters.get(ParameterType.FORWARD_FROM);
+ InetAddressAndPort replyTo;
if (from == null)
{
replyTo = message.from;
- byte[] forwardBytes = message.parameters.get(Mutation.FORWARD_TO);
- if (forwardBytes != null)
- forwardToLocalNodes(message.payload, message.verb, forwardBytes, message.from);
+ ForwardToContainer forwardTo = (ForwardToContainer)message.parameters.get(ParameterType.FORWARD_TO);
+ if (forwardTo != null)
+ forwardToLocalNodes(message.payload, message.verb, forwardTo, message.from);
}
else
{
- replyTo = InetAddress.getByAddress(from);
+
+ replyTo = from;
}
try
@@ -69,22 +69,17 @@ public class MutationVerbHandler implements IVerbHandler
}
}
- private static void forwardToLocalNodes(Mutation mutation, MessagingService.Verb verb, byte[] forwardBytes, InetAddress from) throws IOException
+ private static void forwardToLocalNodes(Mutation mutation, MessagingService.Verb verb, ForwardToContainer forwardTo, InetAddressAndPort from) throws IOException
{
- try (DataInputStream in = new DataInputStream(new FastByteArrayInputStream(forwardBytes)))
+ // tell the recipients who to send their ack to
+ MessageOut message = new MessageOut<>(verb, mutation, Mutation.serializer).withParameter(ParameterType.FORWARD_FROM, from);
+ Iterator iterator = forwardTo.targets.iterator();
+ // Send a message to each of the addresses on our Forward List
+ for (int i = 0; i < forwardTo.targets.size(); i++)
{
- int size = in.readInt();
-
- // tell the recipients who to send their ack to
- MessageOut message = new MessageOut<>(verb, mutation, Mutation.serializer).withParameter(Mutation.FORWARD_FROM, from.getAddress());
- // Send a message to each of the addresses on our Forward List
- for (int i = 0; i < size; i++)
- {
- InetAddress address = CompactEndpointSerializationHelper.deserialize(in);
- int id = in.readInt();
- Tracing.trace("Enqueuing forwarded write to {}", address);
- MessagingService.instance().sendOneWay(message, id, address);
- }
+ InetAddressAndPort address = iterator.next();
+ Tracing.trace("Enqueuing forwarded write to {}", address);
+ MessagingService.instance().sendOneWay(message, forwardTo.messageIds[i], address);
}
}
}
diff --git a/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java b/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
index 066b2fe178..151e7d3202 100644
--- a/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
+++ b/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
@@ -60,7 +60,7 @@ public class SizeEstimatesRecorder extends SchemaChangeListener implements Runna
public void run()
{
TokenMetadata metadata = StorageService.instance.getTokenMetadata().cloneOnlyTokenMap();
- if (!metadata.isMember(FBUtilities.getBroadcastAddress()))
+ if (!metadata.isMember(FBUtilities.getBroadcastAddressAndPort()))
{
logger.debug("Node is not part of the ring; not recording size estimates");
return;
@@ -71,7 +71,7 @@ public class SizeEstimatesRecorder extends SchemaChangeListener implements Runna
for (Keyspace keyspace : Keyspace.nonLocalStrategy())
{
Collection> localRanges = StorageService.instance.getPrimaryRangesForEndpoint(keyspace.getName(),
- FBUtilities.getBroadcastAddress());
+ FBUtilities.getBroadcastAddressAndPort());
for (ColumnFamilyStore table : keyspace.getColumnFamilyStores())
{
long start = System.nanoTime();
diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java
index 4469384cec..91a1bffc97 100644
--- a/src/java/org/apache/cassandra/db/SystemKeyspace.java
+++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java
@@ -35,6 +35,7 @@ import com.google.common.collect.ImmutableSet;
import com.google.common.collect.SetMultimap;
import com.google.common.collect.Sets;
import com.google.common.io.ByteStreams;
+import com.google.common.util.concurrent.ListenableFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -52,6 +53,7 @@ import org.apache.cassandra.dht.*;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.util.*;
import org.apache.cassandra.locator.IEndpointSnitch;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.metrics.RestorableMeter;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.*;
@@ -91,47 +93,52 @@ public final class SystemKeyspace
public static final String PAXOS = "paxos";
public static final String BUILT_INDEXES = "IndexInfo";
public static final String LOCAL = "local";
- public static final String PEERS = "peers";
- public static final String PEER_EVENTS = "peer_events";
+ public static final String PEERS_V2 = "peers_v2";
+ public static final String PEER_EVENTS_V2 = "peer_events_v2";
public static final String RANGE_XFERS = "range_xfers";
public static final String COMPACTION_HISTORY = "compaction_history";
public static final String SSTABLE_ACTIVITY = "sstable_activity";
public static final String SIZE_ESTIMATES = "size_estimates";
public static final String AVAILABLE_RANGES = "available_ranges";
public static final String TRANSFERRED_RANGES = "transferred_ranges";
+ public static final String TRANSFERRED_RANGES_V2 = "transferred_ranges_v2";
public static final String VIEW_BUILDS_IN_PROGRESS = "view_builds_in_progress";
public static final String BUILT_VIEWS = "built_views";
public static final String PREPARED_STATEMENTS = "prepared_statements";
public static final String REPAIRS = "repairs";
+ @Deprecated public static final String LEGACY_PEERS = "peers";
+ @Deprecated public static final String LEGACY_PEER_EVENTS = "peer_events";
+ @Deprecated public static final String LEGACY_TRANSFERRED_RANGES = "transferred_ranges";
+
public static final TableMetadata Batches =
parse(BATCHES,
- "batches awaiting replay",
- "CREATE TABLE %s ("
- + "id timeuuid,"
- + "mutations list,"
- + "version int,"
- + "PRIMARY KEY ((id)))")
- .partitioner(new LocalPartitioner(TimeUUIDType.instance))
- .compaction(CompactionParams.scts(singletonMap("min_threshold", "2")))
- .build();
+ "batches awaiting replay",
+ "CREATE TABLE %s ("
+ + "id timeuuid,"
+ + "mutations list,"
+ + "version int,"
+ + "PRIMARY KEY ((id)))")
+ .partitioner(new LocalPartitioner(TimeUUIDType.instance))
+ .compaction(CompactionParams.scts(singletonMap("min_threshold", "2")))
+ .build();
private static final TableMetadata Paxos =
parse(PAXOS,
- "in-progress paxos proposals",
- "CREATE TABLE %s ("
- + "row_key blob,"
- + "cf_id UUID,"
- + "in_progress_ballot timeuuid,"
- + "most_recent_commit blob,"
- + "most_recent_commit_at timeuuid,"
- + "most_recent_commit_version int,"
- + "proposal blob,"
- + "proposal_ballot timeuuid,"
- + "proposal_version int,"
- + "PRIMARY KEY ((row_key), cf_id))")
- .compaction(CompactionParams.lcs(emptyMap()))
- .build();
+ "in-progress paxos proposals",
+ "CREATE TABLE %s ("
+ + "row_key blob,"
+ + "cf_id UUID,"
+ + "in_progress_ballot timeuuid,"
+ + "most_recent_commit blob,"
+ + "most_recent_commit_at timeuuid,"
+ + "most_recent_commit_version int,"
+ + "proposal blob,"
+ + "proposal_ballot timeuuid,"
+ + "proposal_version int,"
+ + "PRIMARY KEY ((row_key), cf_id))")
+ .compaction(CompactionParams.lcs(emptyMap()))
+ .build();
private static final TableMetadata BuiltIndexes =
parse(BUILT_INDEXES,
@@ -145,122 +152,130 @@ public final class SystemKeyspace
private static final TableMetadata Local =
parse(LOCAL,
- "information about the local node",
- "CREATE TABLE %s ("
- + "key text,"
- + "bootstrapped text,"
- + "broadcast_address inet,"
- + "cluster_name text,"
- + "cql_version text,"
- + "data_center text,"
- + "gossip_generation int,"
- + "host_id uuid,"
- + "listen_address inet,"
- + "native_protocol_version text,"
- + "partitioner text,"
- + "rack text,"
- + "release_version text,"
- + "rpc_address inet,"
- + "schema_version uuid,"
- + "tokens set,"
- + "truncated_at map,"
- + "PRIMARY KEY ((key)))")
- .recordDeprecatedSystemColumn("thrift_version", UTF8Type.instance)
- .build();
+ "information about the local node",
+ "CREATE TABLE %s ("
+ + "key text,"
+ + "bootstrapped text,"
+ + "broadcast_address inet,"
+ + "broadcast_port int,"
+ + "cluster_name text,"
+ + "cql_version text,"
+ + "data_center text,"
+ + "gossip_generation int,"
+ + "host_id uuid,"
+ + "listen_address inet,"
+ + "listen_port int,"
+ + "native_protocol_version text,"
+ + "partitioner text,"
+ + "rack text,"
+ + "release_version text,"
+ + "rpc_address inet,"
+ + "rpc_port int,"
+ + "schema_version uuid,"
+ + "tokens set,"
+ + "truncated_at map,"
+ + "PRIMARY KEY ((key)))"
+ ).recordDeprecatedSystemColumn("thrift_version", UTF8Type.instance)
+ .build();
- private static final TableMetadata Peers =
- parse(PEERS,
- "information about known peers in the cluster",
- "CREATE TABLE %s ("
- + "peer inet,"
- + "data_center text,"
- + "host_id uuid,"
- + "preferred_ip inet,"
- + "rack text,"
- + "release_version text,"
- + "rpc_address inet,"
- + "schema_version uuid,"
- + "tokens set,"
- + "PRIMARY KEY ((peer)))")
- .build();
+ private static final TableMetadata PeersV2 =
+ parse(PEERS_V2,
+ "information about known peers in the cluster",
+ "CREATE TABLE %s ("
+ + "peer inet,"
+ + "peer_port int,"
+ + "data_center text,"
+ + "host_id uuid,"
+ + "preferred_ip inet,"
+ + "preferred_port int,"
+ + "rack text,"
+ + "release_version text,"
+ + "native_address inet,"
+ + "native_port int,"
+ + "schema_version uuid,"
+ + "tokens set,"
+ + "PRIMARY KEY ((peer), peer_port))")
+ .build();
- private static final TableMetadata PeerEvents =
- parse(PEER_EVENTS,
- "events related to peers",
- "CREATE TABLE %s ("
- + "peer inet,"
- + "hints_dropped map,"
- + "PRIMARY KEY ((peer)))")
- .build();
+ private static final TableMetadata PeerEventsV2 =
+ parse(PEER_EVENTS_V2,
+ "events related to peers",
+ "CREATE TABLE %s ("
+ + "peer inet,"
+ + "peer_port int,"
+ + "hints_dropped map,"
+ + "PRIMARY KEY ((peer), peer_port))")
+ .build();
private static final TableMetadata RangeXfers =
parse(RANGE_XFERS,
- "ranges requested for transfer",
- "CREATE TABLE %s ("
- + "token_bytes blob,"
- + "requested_at timestamp,"
- + "PRIMARY KEY ((token_bytes)))")
- .build();
+ "ranges requested for transfer",
+ "CREATE TABLE %s ("
+ + "token_bytes blob,"
+ + "requested_at timestamp,"
+ + "PRIMARY KEY ((token_bytes)))")
+ .build();
private static final TableMetadata CompactionHistory =
parse(COMPACTION_HISTORY,
- "week-long compaction history",
- "CREATE TABLE %s ("
- + "id uuid,"
- + "bytes_in bigint,"
- + "bytes_out bigint,"
- + "columnfamily_name text,"
- + "compacted_at timestamp,"
- + "keyspace_name text,"
- + "rows_merged map,"
- + "PRIMARY KEY ((id)))")
- .defaultTimeToLive((int) TimeUnit.DAYS.toSeconds(7))
- .build();
+ "week-long compaction history",
+ "CREATE TABLE %s ("
+ + "id uuid,"
+ + "bytes_in bigint,"
+ + "bytes_out bigint,"
+ + "columnfamily_name text,"
+ + "compacted_at timestamp,"
+ + "keyspace_name text,"
+ + "rows_merged map,"
+ + "PRIMARY KEY ((id)))")
+ .defaultTimeToLive((int) TimeUnit.DAYS.toSeconds(7))
+ .build();
private static final TableMetadata SSTableActivity =
parse(SSTABLE_ACTIVITY,
- "historic sstable read rates",
- "CREATE TABLE %s ("
- + "keyspace_name text,"
- + "columnfamily_name text,"
- + "generation int,"
- + "rate_120m double,"
- + "rate_15m double,"
- + "PRIMARY KEY ((keyspace_name, columnfamily_name, generation)))")
- .build();
+ "historic sstable read rates",
+ "CREATE TABLE %s ("
+ + "keyspace_name text,"
+ + "columnfamily_name text,"
+ + "generation int,"
+ + "rate_120m double,"
+ + "rate_15m double,"
+ + "PRIMARY KEY ((keyspace_name, columnfamily_name, generation)))")
+ .build();
private static final TableMetadata SizeEstimates =
parse(SIZE_ESTIMATES,
- "per-table primary range size estimates",
- "CREATE TABLE %s ("
- + "keyspace_name text,"
- + "table_name text,"
- + "range_start text,"
- + "range_end text,"
- + "mean_partition_size bigint,"
- + "partitions_count bigint,"
- + "PRIMARY KEY ((keyspace_name), table_name, range_start, range_end))")
- .build();
+ "per-table primary range size estimates",
+ "CREATE TABLE %s ("
+ + "keyspace_name text,"
+ + "table_name text,"
+ + "range_start text,"
+ + "range_end text,"
+ + "mean_partition_size bigint,"
+ + "partitions_count bigint,"
+ + "PRIMARY KEY ((keyspace_name), table_name, range_start, range_end))")
+ .build();
private static final TableMetadata AvailableRanges =
parse(AVAILABLE_RANGES,
- "available keyspace/ranges during bootstrap/replace that are ready to be served",
- "CREATE TABLE %s ("
- + "keyspace_name text,"
- + "ranges set,"
- + "PRIMARY KEY ((keyspace_name)))")
- .build();
+ "available keyspace/ranges during bootstrap/replace that are ready to be served",
+ "CREATE TABLE %s ("
+ + "keyspace_name text,"
+ + "ranges set,"
+ + "PRIMARY KEY ((keyspace_name)))")
+ .build();
- private static final TableMetadata TransferredRanges =
- parse(TRANSFERRED_RANGES,
- "record of transferred ranges for streaming operation",
- "CREATE TABLE %s ("
- + "operation text,"
- + "peer inet,"
- + "keyspace_name text,"
- + "ranges set,"
- + "PRIMARY KEY ((operation, keyspace_name), peer))")
- .build();
+ private static final TableMetadata TransferredRangesV2 =
+ parse(TRANSFERRED_RANGES_V2,
+ "record of transferred ranges for streaming operation",
+ "CREATE TABLE %s ("
+ + "operation text,"
+ + "peer inet,"
+ + "peer_port int,"
+ + "keyspace_name text,"
+ + "ranges set,"
+ + "PRIMARY KEY ((operation, keyspace_name), peer, peer_port))")
+ .build();
private static final TableMetadata ViewBuildsInProgress =
parse(VIEW_BUILDS_IN_PROGRESS,
@@ -277,38 +292,79 @@ public final class SystemKeyspace
private static final TableMetadata BuiltViews =
parse(BUILT_VIEWS,
- "built views",
- "CREATE TABLE %s ("
- + "keyspace_name text,"
- + "view_name text,"
- + "status_replicated boolean,"
- + "PRIMARY KEY ((keyspace_name), view_name))")
- .build();
+ "built views",
+ "CREATE TABLE %s ("
+ + "keyspace_name text,"
+ + "view_name text,"
+ + "status_replicated boolean,"
+ + "PRIMARY KEY ((keyspace_name), view_name))")
+ .build();
private static final TableMetadata PreparedStatements =
parse(PREPARED_STATEMENTS,
- "prepared statements",
- "CREATE TABLE %s ("
- + "prepared_id blob,"
- + "logged_keyspace text,"
- + "query_string text,"
- + "PRIMARY KEY ((prepared_id)))")
- .build();
+ "prepared statements",
+ "CREATE TABLE %s ("
+ + "prepared_id blob,"
+ + "logged_keyspace text,"
+ + "query_string text,"
+ + "PRIMARY KEY ((prepared_id)))")
+ .build();
private static final TableMetadata Repairs =
parse(REPAIRS,
- "repairs",
- "CREATE TABLE %s ("
- + "parent_id timeuuid, "
- + "started_at timestamp, "
- + "last_update timestamp, "
- + "repaired_at timestamp, "
- + "state int, "
- + "coordinator inet, "
- + "participants set, "
- + "ranges set, "
- + "cfids set, "
- + "PRIMARY KEY (parent_id))").build();
+ "repairs",
+ "CREATE TABLE %s ("
+ + "parent_id timeuuid, "
+ + "started_at timestamp, "
+ + "last_update timestamp, "
+ + "repaired_at timestamp, "
+ + "state int, "
+ + "coordinator inet, "
+ + "coordinator_port int,"
+ + "participants set,"
+ + "participants_wp set,"
+ + "ranges set, "
+ + "cfids set, "
+ + "PRIMARY KEY (parent_id))").build();
+
+ @Deprecated
+ private static final TableMetadata LegacyPeers =
+ parse(LEGACY_PEERS,
+ "information about known peers in the cluster",
+ "CREATE TABLE %s ("
+ + "peer inet,"
+ + "data_center text,"
+ + "host_id uuid,"
+ + "preferred_ip inet,"
+ + "rack text,"
+ + "release_version text,"
+ + "rpc_address inet,"
+ + "schema_version uuid,"
+ + "tokens set,"
+ + "PRIMARY KEY ((peer)))")
+ .build();
+
+ @Deprecated
+ private static final TableMetadata LegacyPeerEvents =
+ parse(LEGACY_PEER_EVENTS,
+ "events related to peers",
+ "CREATE TABLE %s ("
+ + "peer inet,"
+ + "hints_dropped map,"
+ + "PRIMARY KEY ((peer)))")
+ .build();
+
+ @Deprecated
+ private static final TableMetadata LegacyTransferredRanges =
+ parse(LEGACY_TRANSFERRED_RANGES,
+ "record of transferred ranges for streaming operation",
+ "CREATE TABLE %s ("
+ + "operation text,"
+ + "peer inet,"
+ + "keyspace_name text,"
+ + "ranges set,"
+ + "PRIMARY KEY ((operation, keyspace_name), peer))")
+ .build();
private static TableMetadata.Builder parse(String table, String description, String cql)
{
@@ -331,14 +387,17 @@ public final class SystemKeyspace
Batches,
Paxos,
Local,
- Peers,
- PeerEvents,
+ PeersV2,
+ LegacyPeers,
+ PeerEventsV2,
+ LegacyPeerEvents,
RangeXfers,
CompactionHistory,
SSTableActivity,
SizeEstimates,
AvailableRanges,
- TransferredRanges,
+ TransferredRangesV2,
+ LegacyTransferredRanges,
ViewBuildsInProgress,
BuiltViews,
PreparedStatements,
@@ -384,9 +443,12 @@ public final class SystemKeyspace
"rack," +
"partitioner," +
"rpc_address," +
+ "rpc_port," +
"broadcast_address," +
- "listen_address" +
- ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
+ "broadcast_port," +
+ "listen_address," +
+ "listen_port" +
+ ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
executeOnceInternal(format(req, LOCAL),
LOCAL,
@@ -394,12 +456,15 @@ public final class SystemKeyspace
FBUtilities.getReleaseVersionString(),
QueryProcessor.CQL_VERSION.toString(),
String.valueOf(ProtocolVersion.CURRENT.asInt()),
- snitch.getDatacenter(FBUtilities.getBroadcastAddress()),
- snitch.getRack(FBUtilities.getBroadcastAddress()),
+ snitch.getDatacenter(FBUtilities.getBroadcastAddressAndPort()),
+ snitch.getRack(FBUtilities.getBroadcastAddressAndPort()),
DatabaseDescriptor.getPartitioner().getClass().getName(),
DatabaseDescriptor.getRpcAddress(),
- FBUtilities.getBroadcastAddress(),
- FBUtilities.getLocalAddress());
+ DatabaseDescriptor.getNativeTransportPort(),
+ FBUtilities.getJustBroadcastAddress(),
+ DatabaseDescriptor.getStoragePort(),
+ FBUtilities.getJustLocalAddress(),
+ DatabaseDescriptor.getStoragePort());
}
public static void updateCompactionHistory(String ksname,
@@ -461,11 +526,10 @@ public final class SystemKeyspace
{
String buildReq = "DELETE FROM %s.%s WHERE keyspace_name = ? AND view_name = ?";
executeInternal(String.format(buildReq, SchemaConstants.SYSTEM_KEYSPACE_NAME, VIEW_BUILDS_IN_PROGRESS), keyspaceName, viewName);
- forceBlockingFlush(VIEW_BUILDS_IN_PROGRESS);
String builtReq = "DELETE FROM %s.\"%s\" WHERE keyspace_name = ? AND view_name = ? IF EXISTS";
executeInternal(String.format(builtReq, SchemaConstants.SYSTEM_KEYSPACE_NAME, BUILT_VIEWS), keyspaceName, viewName);
- forceBlockingFlush(BUILT_VIEWS);
+ forceBlockingFlush(VIEW_BUILDS_IN_PROGRESS, BUILT_VIEWS);
}
public static void finishViewBuildStatus(String ksname, String viewName)
@@ -609,39 +673,64 @@ public final class SystemKeyspace
/**
* Record tokens being used by another node
*/
- public static synchronized void updateTokens(InetAddress ep, Collection tokens)
+ public static synchronized void updateTokens(InetAddressAndPort ep, Collection tokens)
{
- if (ep.equals(FBUtilities.getBroadcastAddress()))
+ if (ep.equals(FBUtilities.getBroadcastAddressAndPort()))
return;
String req = "INSERT INTO system.%s (peer, tokens) VALUES (?, ?)";
- executeInternal(format(req, PEERS), ep, tokensAsSet(tokens));
+ executeInternal(String.format(req, LEGACY_PEERS), ep.address, tokensAsSet(tokens));
+ req = "INSERT INTO system.%s (peer, peer_port, tokens) VALUES (?, ?, ?)";
+ executeInternal(String.format(req, PEERS_V2), ep.address, ep.port, tokensAsSet(tokens));
}
- public static synchronized void updatePreferredIP(InetAddress ep, InetAddress preferred_ip)
+ public static synchronized void updatePreferredIP(InetAddressAndPort ep, InetAddressAndPort preferred_ip)
{
if (getPreferredIP(ep) == preferred_ip)
return;
String req = "INSERT INTO system.%s (peer, preferred_ip) VALUES (?, ?)";
- executeInternal(format(req, PEERS), ep, preferred_ip);
- forceBlockingFlush(PEERS);
+ executeInternal(String.format(req, LEGACY_PEERS), ep.address, preferred_ip.address);
+ req = "INSERT INTO system.%s (peer, peer_port, preferred_ip, preferred_port) VALUES (?, ?, ?, ?)";
+ executeInternal(String.format(req, PEERS_V2), ep.address, ep.port, preferred_ip.address, preferred_ip.port);
+ forceBlockingFlush(LEGACY_PEERS, PEERS_V2);
}
- public static synchronized void updatePeerInfo(InetAddress ep, String columnName, Object value)
+ public static synchronized void updatePeerInfo(InetAddressAndPort ep, String columnName, Object value)
{
- if (ep.equals(FBUtilities.getBroadcastAddress()))
+ if (ep.equals(FBUtilities.getBroadcastAddressAndPort()))
return;
String req = "INSERT INTO system.%s (peer, %s) VALUES (?, ?)";
- executeInternal(format(req, PEERS, columnName), ep, value);
+ executeInternal(String.format(req, LEGACY_PEERS, columnName), ep.address, value);
+ //This column doesn't match across the two tables
+ if (columnName.equals("rpc_address"))
+ {
+ columnName = "native_address";
+ }
+ req = "INSERT INTO system.%s (peer, peer_port, %s) VALUES (?, ?, ?)";
+ executeInternal(String.format(req, PEERS_V2, columnName), ep.address, ep.port, value);
}
- public static synchronized void updateHintsDropped(InetAddress ep, UUID timePeriod, int value)
+ public static synchronized void updatePeerNativeAddress(InetAddressAndPort ep, InetAddressAndPort address)
+ {
+ if (ep.equals(FBUtilities.getBroadcastAddressAndPort()))
+ return;
+
+ String req = "INSERT INTO system.%s (peer, rpc_address) VALUES (?, ?)";
+ executeInternal(String.format(req, LEGACY_PEERS), ep.address, address.address);
+ req = "INSERT INTO system.%s (peer, peer_port, native_address, native_port) VALUES (?, ?, ?, ?)";
+ executeInternal(String.format(req, PEERS_V2), ep.address, ep.port, address.address, address.port);
+ }
+
+
+ public static synchronized void updateHintsDropped(InetAddressAndPort ep, UUID timePeriod, int value)
{
// with 30 day TTL
String req = "UPDATE system.%s USING TTL 2592000 SET hints_dropped[ ? ] = ? WHERE peer = ?";
- executeInternal(format(req, PEER_EVENTS), timePeriod, value, ep);
+ executeInternal(String.format(req, LEGACY_PEER_EVENTS), timePeriod, value, ep.address);
+ req = "UPDATE system.%s USING TTL 2592000 SET hints_dropped[ ? ] = ? WHERE peer = ? AND peer_port = ?";
+ executeInternal(String.format(req, PEER_EVENTS_V2), timePeriod, value, ep.address, ep.port);
}
public static synchronized void updateSchemaVersion(UUID version)
@@ -673,11 +762,13 @@ public final class SystemKeyspace
/**
* Remove stored tokens being used by another node
*/
- public static synchronized void removeEndpoint(InetAddress ep)
+ public static synchronized void removeEndpoint(InetAddressAndPort ep)
{
String req = "DELETE FROM system.%s WHERE peer = ?";
- executeInternal(format(req, PEERS), ep);
- forceBlockingFlush(PEERS);
+ executeInternal(String.format(req, LEGACY_PEERS), ep.address);
+ req = String.format("DELETE FROM system.%s WHERE peer = ? AND peer_port = ?", PEERS_V2);
+ executeInternal(req, ep.address, ep.port);
+ forceBlockingFlush(LEGACY_PEERS, PEERS_V2);
}
/**
@@ -696,22 +787,32 @@ public final class SystemKeyspace
forceBlockingFlush(LOCAL);
}
- public static void forceBlockingFlush(String cfname)
+ public static void forceBlockingFlush(String ...cfnames)
{
if (!DatabaseDescriptor.isUnsafeSystem())
- FBUtilities.waitOnFuture(Keyspace.open(SchemaConstants.SYSTEM_KEYSPACE_NAME).getColumnFamilyStore(cfname).forceFlush());
+ {
+ List> futures = new ArrayList<>();
+
+ for (String cfname : cfnames)
+ {
+ futures.add(Keyspace.open(SchemaConstants.SYSTEM_KEYSPACE_NAME).getColumnFamilyStore(cfname).forceFlush());
+ }
+ FBUtilities.waitOnFutures(futures);
+ }
}
/**
* Return a map of stored tokens to IP addresses
*
*/
- public static SetMultimap loadTokens()
+ public static SetMultimap loadTokens()
{
- SetMultimap tokenMap = HashMultimap.create();
- for (UntypedResultSet.Row row : executeInternal("SELECT peer, tokens FROM system." + PEERS))
+ SetMultimap tokenMap = HashMultimap.create();
+ for (UntypedResultSet.Row row : executeInternal("SELECT peer, peer_port, tokens FROM system." + PEERS_V2))
{
- InetAddress peer = row.getInetAddress("peer");
+ InetAddress address = row.getInetAddress("peer");
+ Integer port = row.getInt("peer_port");
+ InetAddressAndPort peer = InetAddressAndPort.getByAddressOverrideDefaults(address, port);
if (row.has("tokens"))
tokenMap.putAll(peer, deserializeTokens(row.getSet("tokens", UTF8Type.instance)));
}
@@ -723,12 +824,14 @@ public final class SystemKeyspace
* Return a map of store host_ids to IP addresses
*
*/
- public static Map loadHostIds()
+ public static Map loadHostIds()
{
- Map hostIdMap = new HashMap<>();
- for (UntypedResultSet.Row row : executeInternal("SELECT peer, host_id FROM system." + PEERS))
+ Map hostIdMap = new HashMap<>();
+ for (UntypedResultSet.Row row : executeInternal("SELECT peer, peer_port, host_id FROM system." + PEERS_V2))
{
- InetAddress peer = row.getInetAddress("peer");
+ InetAddress address = row.getInetAddress("peer");
+ Integer port = row.getInt("peer_port");
+ InetAddressAndPort peer = InetAddressAndPort.getByAddressOverrideDefaults(address, port);
if (row.has("host_id"))
{
hostIdMap.put(peer, row.getUUID("host_id"));
@@ -743,24 +846,29 @@ public final class SystemKeyspace
* @param ep endpoint address to check
* @return Preferred IP for given endpoint if present, otherwise returns given ep
*/
- public static InetAddress getPreferredIP(InetAddress ep)
+ public static InetAddressAndPort getPreferredIP(InetAddressAndPort ep)
{
- String req = "SELECT preferred_ip FROM system.%s WHERE peer=?";
- UntypedResultSet result = executeInternal(format(req, PEERS), ep);
+ String req = "SELECT preferred_ip, preferred_port FROM system.%s WHERE peer=? AND peer_port = ?";
+ UntypedResultSet result = executeInternal(String.format(req, PEERS_V2), ep.address, ep.port);
if (!result.isEmpty() && result.one().has("preferred_ip"))
- return result.one().getInetAddress("preferred_ip");
+ {
+ UntypedResultSet.Row row = result.one();
+ return InetAddressAndPort.getByAddressOverrideDefaults(row.getInetAddress("preferred_ip"), row.getInt("preferred_port"));
+ }
return ep;
}
/**
* Return a map of IP addresses containing a map of dc and rack info
*/
- public static Map> loadDcRackInfo()
+ public static Map> loadDcRackInfo()
{
- Map> result = new HashMap<>();
- for (UntypedResultSet.Row row : executeInternal("SELECT peer, data_center, rack from system." + PEERS))
+ Map> result = new HashMap<>();
+ for (UntypedResultSet.Row row : executeInternal("SELECT peer, peer_port, data_center, rack from system." + PEERS_V2))
{
- InetAddress peer = row.getInetAddress("peer");
+ InetAddress address = row.getInetAddress("peer");
+ Integer port = row.getInt("peer_port");
+ InetAddressAndPort peer = InetAddressAndPort.getByAddressOverrideDefaults(address, port);
if (row.has("data_center") && row.has("rack"))
{
Map dcRack = new HashMap<>();
@@ -779,16 +887,16 @@ public final class SystemKeyspace
* @param ep endpoint address to check
* @return Release version or null if version is unknown.
*/
- public static CassandraVersion getReleaseVersion(InetAddress ep)
+ public static CassandraVersion getReleaseVersion(InetAddressAndPort ep)
{
try
{
- if (FBUtilities.getBroadcastAddress().equals(ep))
+ if (FBUtilities.getBroadcastAddressAndPort().equals(ep))
{
return new CassandraVersion(FBUtilities.getReleaseVersionString());
}
- String req = "SELECT release_version FROM system.%s WHERE peer=?";
- UntypedResultSet result = executeInternal(format(req, PEERS), ep);
+ String req = "SELECT release_version FROM system.%s WHERE peer=? AND peer_port=?";
+ UntypedResultSet result = executeInternal(String.format(req, PEERS_V2), ep.address, ep.port);
if (result != null && result.one().has("release_version"))
{
return new CassandraVersion(result.one().getString("release_version"));
@@ -1197,7 +1305,7 @@ public final class SystemKeyspace
}
public static synchronized void updateTransferredRanges(StreamOperation streamOperation,
- InetAddress peer,
+ InetAddressAndPort peer,
String keyspace,
Collection> streamedRanges)
{
@@ -1207,17 +1315,21 @@ public final class SystemKeyspace
{
rangesToUpdate.add(rangeToBytes(range));
}
- executeInternal(format(cql, TRANSFERRED_RANGES), rangesToUpdate, streamOperation.getDescription(), peer, keyspace);
+ executeInternal(format(cql, LEGACY_TRANSFERRED_RANGES), rangesToUpdate, streamOperation.getDescription(), peer.address, keyspace);
+ cql = "UPDATE system.%s SET ranges = ranges + ? WHERE operation = ? AND peer = ? AND peer_port = ? AND keyspace_name = ?";
+ executeInternal(String.format(cql, TRANSFERRED_RANGES_V2), rangesToUpdate, streamOperation.getDescription(), peer.address, peer.port, keyspace);
}
- public static synchronized Map>> getTransferredRanges(String description, String keyspace, IPartitioner partitioner)
+ public static synchronized Map>> getTransferredRanges(String description, String keyspace, IPartitioner partitioner)
{
- Map>> result = new HashMap<>();
+ Map>> result = new HashMap<>();
String query = "SELECT * FROM system.%s WHERE operation = ? AND keyspace_name = ?";
- UntypedResultSet rs = executeInternal(format(query, TRANSFERRED_RANGES), description, keyspace);
+ UntypedResultSet rs = executeInternal(String.format(query, TRANSFERRED_RANGES_V2), description, keyspace);
for (UntypedResultSet.Row row : rs)
{
- InetAddress peer = row.getInetAddress("peer");
+ InetAddress peerAddress = row.getInetAddress("peer");
+ int port = row.getInt("peer_port");
+ InetAddressAndPort peer = InetAddressAndPort.getByAddressOverrideDefaults(peerAddress, port);
Set rawRanges = row.getSet("ranges", BytesType.instance);
Set> ranges = Sets.newHashSetWithExpectedSize(rawRanges.size());
for (ByteBuffer rawRange : rawRanges)
diff --git a/src/java/org/apache/cassandra/db/SystemKeyspaceMigrator40.java b/src/java/org/apache/cassandra/db/SystemKeyspaceMigrator40.java
new file mode 100644
index 0000000000..ea5ff59b04
--- /dev/null
+++ b/src/java/org/apache/cassandra/db/SystemKeyspaceMigrator40.java
@@ -0,0 +1,184 @@
+/*
+ * 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.db;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.schema.SchemaConstants;
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.marshal.BytesType;
+import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.db.marshal.UUIDType;
+
+/**
+ * Migrate 3.0 versions of some tables to 4.0. In this case it's just extra columns and some keys
+ * that are changed.
+ *
+ * Can't just add the additional columns because they are primary key columns and C* doesn't support changing
+ * key columns even if it's just clustering columns.
+ */
+public class SystemKeyspaceMigrator40
+{
+ static final String legacyPeersName = String.format("%s.%s", SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.LEGACY_PEERS);
+ static final String peersName = String.format("%s.%s", SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.PEERS_V2);
+ static final String legacyPeerEventsName = String.format("%s.%s", SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.LEGACY_PEER_EVENTS);
+ static final String peerEventsName = String.format("%s.%s", SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.PEER_EVENTS_V2);
+ static final String legacyTransferredRangesName = String.format("%s.%s", SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.LEGACY_TRANSFERRED_RANGES);
+ static final String transferredRangesName = String.format("%s.%s", SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.TRANSFERRED_RANGES_V2);
+
+ private static final Logger logger = LoggerFactory.getLogger(SystemKeyspaceMigrator40.class);
+
+ private SystemKeyspaceMigrator40() {}
+
+ public static void migrate()
+ {
+ migratePeers();
+ migratePeerEvents();
+ migrateTransferredRanges();
+ }
+
+ private static void migratePeers()
+ {
+ ColumnFamilyStore newPeers = Keyspace.open(SchemaConstants.SYSTEM_KEYSPACE_NAME).getColumnFamilyStore(SystemKeyspace.PEERS_V2);
+
+ if (!newPeers.isEmpty())
+ return;
+
+ logger.info("{} table was empty, migrating legacy {}, if this fails you should fix the issue and then truncate {} to have it try again.",
+ peersName, legacyPeersName, peersName);
+
+ String query = String.format("SELECT * FROM %s",
+ legacyPeersName);
+
+ String insert = String.format("INSERT INTO %s ( "
+ + "peer, "
+ + "peer_port, "
+ + "data_center, "
+ + "host_id, "
+ + "preferred_ip, "
+ + "preferred_port, "
+ + "rack, "
+ + "release_version, "
+ + "native_address, "
+ + "native_port, "
+ + "schema_version, "
+ + "tokens) "
+ + " values ( ?, ?, ? , ? , ?, ?, ?, ?, ?, ?, ?, ?)",
+ peersName);
+
+ UntypedResultSet rows = QueryProcessor.executeInternalWithPaging(query, 1000);
+ int transferred = 0;
+ logger.info("Migrating rows from legacy {} to {}", legacyPeersName, peersName);
+ for (UntypedResultSet.Row row : rows)
+ {
+ logger.debug("Transferring row {}", transferred);
+ QueryProcessor.executeInternal(insert,
+ row.has("peer") ? row.getInetAddress("peer") : null,
+ DatabaseDescriptor.getStoragePort(),
+ row.has("data_center") ? row.getString("data_center") : null,
+ row.has("host_id") ? row.getUUID("host_id") : null,
+ row.has("preferred_ip") ? row.getInetAddress("preferred_ip") : null,
+ DatabaseDescriptor.getStoragePort(),
+ row.has("rack") ? row.getString("rack") : null,
+ row.has("release_version") ? row.getString("release_version") : null,
+ row.has("rpc_address") ? row.getInetAddress("rpc_address") : null,
+ DatabaseDescriptor.getNativeTransportPort(),
+ row.has("schema_version") ? row.getUUID("schema_version") : null,
+ row.has("tokens") ? row.getSet("tokens", UTF8Type.instance) : null);
+ transferred++;
+ }
+ logger.info("Migrated {} rows from legacy {} to {}", transferred, legacyPeersName, peersName);
+ }
+
+ private static void migratePeerEvents()
+ {
+ ColumnFamilyStore newPeerEvents = Keyspace.open(SchemaConstants.SYSTEM_KEYSPACE_NAME).getColumnFamilyStore(SystemKeyspace.PEER_EVENTS_V2);
+
+ if (!newPeerEvents.isEmpty())
+ return;
+
+ logger.info("{} table was empty, migrating legacy {} to {}", peerEventsName, legacyPeerEventsName, peerEventsName);
+
+ String query = String.format("SELECT * FROM %s",
+ legacyPeerEventsName);
+
+ String insert = String.format("INSERT INTO %s ( "
+ + "peer, "
+ + "peer_port, "
+ + "hints_dropped) "
+ + " values ( ?, ?, ? )",
+ peerEventsName);
+
+ UntypedResultSet rows = QueryProcessor.executeInternalWithPaging(query, 1000);
+ int transferred = 0;
+ for (UntypedResultSet.Row row : rows)
+ {
+ logger.debug("Transferring row {}", transferred);
+ QueryProcessor.executeInternal(insert,
+ row.has("peer") ? row.getInetAddress("peer") : null,
+ DatabaseDescriptor.getStoragePort(),
+ row.has("hints_dropped") ? row.getMap("hints_dropped", UUIDType.instance, Int32Type.instance) : null);
+ transferred++;
+ }
+ logger.info("Migrated {} rows from legacy {} to {}", transferred, legacyPeerEventsName, peerEventsName);
+ }
+
+ static void migrateTransferredRanges()
+ {
+ ColumnFamilyStore newTransferredRanges = Keyspace.open(SchemaConstants.SYSTEM_KEYSPACE_NAME).getColumnFamilyStore(SystemKeyspace.TRANSFERRED_RANGES_V2);
+
+ if (!newTransferredRanges.isEmpty())
+ return;
+
+ logger.info("{} table was empty, migrating legacy {} to {}", transferredRangesName, legacyTransferredRangesName, transferredRangesName);
+
+ String query = String.format("SELECT * FROM %s",
+ legacyTransferredRangesName);
+
+ String insert = String.format("INSERT INTO %s ("
+ + "operation, "
+ + "peer, "
+ + "peer_port, "
+ + "keyspace_name, "
+ + "ranges) "
+ + " values ( ?, ?, ? , ?, ?)",
+ transferredRangesName);
+
+ UntypedResultSet rows = QueryProcessor.executeInternalWithPaging(query, 1000);
+ int transferred = 0;
+ for (UntypedResultSet.Row row : rows)
+ {
+ logger.debug("Transferring row {}", transferred);
+ QueryProcessor.executeInternal(insert,
+ row.has("operation") ? row.getString("operation") : null,
+ row.has("peer") ? row.getInetAddress("peer") : null,
+ DatabaseDescriptor.getStoragePort(),
+ row.has("keyspace_name") ? row.getString("keyspace_name") : null,
+ row.has("ranges") ? row.getSet("ranges", BytesType.instance) : null);
+ transferred++;
+ }
+
+ logger.info("Migrated {} rows from legacy {} to {}", transferred, legacyTransferredRangesName, transferredRangesName);
+ }
+
+}
diff --git a/src/java/org/apache/cassandra/db/view/ViewUtils.java b/src/java/org/apache/cassandra/db/view/ViewUtils.java
index 4dc17662c3..df16943263 100644
--- a/src/java/org/apache/cassandra/db/view/ViewUtils.java
+++ b/src/java/org/apache/cassandra/db/view/ViewUtils.java
@@ -18,7 +18,6 @@
package org.apache.cassandra.db.view;
-import java.net.InetAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
@@ -27,6 +26,7 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.NetworkTopologyStrategy;
import org.apache.cassandra.utils.FBUtilities;
@@ -58,14 +58,14 @@ public final class ViewUtils
*
* @return Optional.empty() if this method is called using a base token which does not belong to this replica
*/
- public static Optional getViewNaturalEndpoint(String keyspaceName, Token baseToken, Token viewToken)
+ public static Optional getViewNaturalEndpoint(String keyspaceName, Token baseToken, Token viewToken)
{
AbstractReplicationStrategy replicationStrategy = Keyspace.open(keyspaceName).getReplicationStrategy();
- String localDataCenter = DatabaseDescriptor.getEndpointSnitch().getDatacenter(FBUtilities.getBroadcastAddress());
- List baseEndpoints = new ArrayList<>();
- List viewEndpoints = new ArrayList<>();
- for (InetAddress baseEndpoint : replicationStrategy.getNaturalEndpoints(baseToken))
+ String localDataCenter = DatabaseDescriptor.getEndpointSnitch().getDatacenter(FBUtilities.getBroadcastAddressAndPort());
+ List baseEndpoints = new ArrayList<>();
+ List viewEndpoints = new ArrayList<>();
+ for (InetAddressAndPort baseEndpoint : replicationStrategy.getNaturalEndpoints(baseToken))
{
// An endpoint is local if we're not using Net
if (!(replicationStrategy instanceof NetworkTopologyStrategy) ||
@@ -73,10 +73,10 @@ public final class ViewUtils
baseEndpoints.add(baseEndpoint);
}
- for (InetAddress viewEndpoint : replicationStrategy.getNaturalEndpoints(viewToken))
+ for (InetAddressAndPort viewEndpoint : replicationStrategy.getNaturalEndpoints(viewToken))
{
// If we are a base endpoint which is also a view replica, we use ourselves as our view replica
- if (viewEndpoint.equals(FBUtilities.getBroadcastAddress()))
+ if (viewEndpoint.equals(FBUtilities.getBroadcastAddressAndPort()))
return Optional.of(viewEndpoint);
// We have to remove any endpoint which is shared between the base and the view, as it will select itself
@@ -92,7 +92,7 @@ public final class ViewUtils
// Since the same replication strategy is used, the same placement should be used and we should get the same
// number of replicas for all of the tokens in the ring.
assert baseEndpoints.size() == viewEndpoints.size() : "Replication strategy should have the same number of endpoints for the base and the view";
- int baseIdx = baseEndpoints.indexOf(FBUtilities.getBroadcastAddress());
+ int baseIdx = baseEndpoints.indexOf(FBUtilities.getBroadcastAddressAndPort());
if (baseIdx < 0)
//This node is not a base replica of this key, so we return empty
diff --git a/src/java/org/apache/cassandra/dht/BootStrapper.java b/src/java/org/apache/cassandra/dht/BootStrapper.java
index a25f867d5e..432586b13b 100644
--- a/src/java/org/apache/cassandra/dht/BootStrapper.java
+++ b/src/java/org/apache/cassandra/dht/BootStrapper.java
@@ -18,7 +18,6 @@
package org.apache.cassandra.dht;
import java.io.IOException;
-import java.net.InetAddress;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
@@ -38,6 +37,7 @@ import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.streaming.*;
@@ -51,12 +51,12 @@ public class BootStrapper extends ProgressEventNotifierSupport
private static final Logger logger = LoggerFactory.getLogger(BootStrapper.class);
/* endpoint that needs to be bootstrapped */
- protected final InetAddress address;
+ protected final InetAddressAndPort address;
/* token of the node being bootstrapped. */
protected final Collection tokens;
protected final TokenMetadata tokenMetadata;
- public BootStrapper(InetAddress address, Collection tokens, TokenMetadata tmd)
+ public BootStrapper(InetAddressAndPort address, Collection tokens, TokenMetadata tmd)
{
assert address != null;
assert tokens != null && !tokens.isEmpty();
@@ -159,7 +159,7 @@ public class BootStrapper extends ProgressEventNotifierSupport
* otherwise, if allocationKeyspace is specified use the token allocation algorithm to generate suitable tokens
* else choose num_tokens tokens at random
*/
- public static Collection getBootstrapTokens(final TokenMetadata metadata, InetAddress address, int schemaWaitDelay) throws ConfigurationException
+ public static Collection getBootstrapTokens(final TokenMetadata metadata, InetAddressAndPort address, int schemaWaitDelay) throws ConfigurationException
{
String allocationKeyspace = DatabaseDescriptor.getAllocateTokensForKeyspace();
Collection initialTokens = DatabaseDescriptor.getInitialTokens();
@@ -199,13 +199,13 @@ public class BootStrapper extends ProgressEventNotifierSupport
}
static Collection allocateTokens(final TokenMetadata metadata,
- InetAddress address,
+ InetAddressAndPort address,
String allocationKeyspace,
int numTokens,
int schemaWaitDelay)
{
StorageService.instance.waitForSchema(schemaWaitDelay);
- if (!FBUtilities.getBroadcastAddress().equals(InetAddress.getLoopbackAddress()))
+ if (!FBUtilities.getBroadcastAddressAndPort().equals(InetAddressAndPort.getLoopbackAddress()))
Gossiper.waitToSettle();
Keyspace ks = Keyspace.open(allocationKeyspace);
diff --git a/src/java/org/apache/cassandra/dht/RangeFetchMapCalculator.java b/src/java/org/apache/cassandra/dht/RangeFetchMapCalculator.java
index d407212989..b90bc96392 100644
--- a/src/java/org/apache/cassandra/dht/RangeFetchMapCalculator.java
+++ b/src/java/org/apache/cassandra/dht/RangeFetchMapCalculator.java
@@ -32,6 +32,7 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.utils.FBUtilities;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -72,7 +73,7 @@ public class RangeFetchMapCalculator
{
private static final Logger logger = LoggerFactory.getLogger(RangeFetchMapCalculator.class);
private static final long TRIVIAL_RANGE_LIMIT = 1000;
- private final Multimap, InetAddress> rangesWithSources;
+ private final Multimap, InetAddressAndPort> rangesWithSources;
private final Collection sourceFilters;
private final String keyspace;
//We need two Vertices to act as source and destination in the algorithm
@@ -80,7 +81,7 @@ public class RangeFetchMapCalculator
private final Vertex destinationVertex = OuterVertex.getDestinationVertex();
private final Set> trivialRanges;
- public RangeFetchMapCalculator(Multimap, InetAddress> rangesWithSources,
+ public RangeFetchMapCalculator(Multimap, InetAddressAndPort> rangesWithSources,
Collection sourceFilters,
String keyspace)
{
@@ -108,16 +109,16 @@ public class RangeFetchMapCalculator
return false;
}
- public Multimap> getRangeFetchMap()
+ public Multimap> getRangeFetchMap()
{
- Multimap> fetchMap = HashMultimap.create();
+ Multimap> fetchMap = HashMultimap.create();
fetchMap.putAll(getRangeFetchMapForNonTrivialRanges());
fetchMap.putAll(getRangeFetchMapForTrivialRanges(fetchMap));
return fetchMap;
}
@VisibleForTesting
- Multimap> getRangeFetchMapForNonTrivialRanges()
+ Multimap> getRangeFetchMapForNonTrivialRanges()
{
//Get the graph with edges between ranges and their source endpoints
MutableCapacityGraph graph = getGraph();
@@ -148,19 +149,19 @@ public class RangeFetchMapCalculator
}
@VisibleForTesting
- Multimap> getRangeFetchMapForTrivialRanges(Multimap> optimisedMap)
+ Multimap> getRangeFetchMapForTrivialRanges(Multimap> optimisedMap)
{
- Multimap> fetchMap = HashMultimap.create();
+ Multimap> fetchMap = HashMultimap.create();
for (Range trivialRange : trivialRanges)
{
boolean added = false;
boolean localDCCheck = true;
while (!added)
{
- List srcs = new ArrayList<>(rangesWithSources.get(trivialRange));
+ List srcs = new ArrayList<>(rangesWithSources.get(trivialRange));
// sort with the endpoint having the least number of streams first:
srcs.sort(Comparator.comparingInt(o -> optimisedMap.get(o).size()));
- for (InetAddress src : srcs)
+ for (InetAddressAndPort src : srcs)
{
if (passFilters(src, localDCCheck))
{
@@ -202,9 +203,9 @@ public class RangeFetchMapCalculator
* @param result Flow algorithm result
* @return Multi Map of Machine to Ranges
*/
- private Multimap> getRangeFetchMapFromGraphResult(MutableCapacityGraph graph, MaximumFlowAlgorithmResult> result)
+ private Multimap> getRangeFetchMapFromGraphResult(MutableCapacityGraph graph, MaximumFlowAlgorithmResult> result)
{
- final Multimap> rangeFetchMapMap = HashMultimap.create();
+ final Multimap> rangeFetchMapMap = HashMultimap.create();
if(result == null)
return rangeFetchMapMap;
final Function, Integer> flowFunction = result.calcFlowFunction();
@@ -346,13 +347,13 @@ public class RangeFetchMapCalculator
private boolean addEndpoints(MutableCapacityGraph capacityGraph, RangeVertex rangeVertex, boolean localDCCheck)
{
boolean sourceFound = false;
- for (InetAddress endpoint : rangesWithSources.get(rangeVertex.getRange()))
+ for (InetAddressAndPort endpoint : rangesWithSources.get(rangeVertex.getRange()))
{
if (passFilters(endpoint, localDCCheck))
{
sourceFound = true;
// if we pass filters, it means that we don't filter away localhost and we can count it as a source:
- if (endpoint.equals(FBUtilities.getBroadcastAddress()))
+ if (endpoint.equals(FBUtilities.getBroadcastAddressAndPort()))
continue; // but don't add localhost to the graph to avoid streaming locally
final Vertex endpointVertex = new EndpointVertex(endpoint);
capacityGraph.insertVertex(rangeVertex);
@@ -363,7 +364,7 @@ public class RangeFetchMapCalculator
return sourceFound;
}
- private boolean isInLocalDC(InetAddress endpoint)
+ private boolean isInLocalDC(InetAddressAndPort endpoint)
{
return DatabaseDescriptor.getLocalDataCenter().equals(DatabaseDescriptor.getEndpointSnitch().getDatacenter(endpoint));
}
@@ -374,7 +375,7 @@ public class RangeFetchMapCalculator
* @param localDCCheck Allow endpoints with local DC
* @return True if filters pass this endpoint
*/
- private boolean passFilters(final InetAddress endpoint, boolean localDCCheck)
+ private boolean passFilters(final InetAddressAndPort endpoint, boolean localDCCheck)
{
for (RangeStreamer.ISourceFilter filter : sourceFilters)
{
@@ -410,15 +411,15 @@ public class RangeFetchMapCalculator
*/
private static class EndpointVertex extends Vertex
{
- private final InetAddress endpoint;
+ private final InetAddressAndPort endpoint;
- public EndpointVertex(InetAddress endpoint)
+ public EndpointVertex(InetAddressAndPort endpoint)
{
assert endpoint != null;
this.endpoint = endpoint;
}
- public InetAddress getEndpoint()
+ public InetAddressAndPort getEndpoint()
{
return endpoint;
}
diff --git a/src/java/org/apache/cassandra/dht/RangeStreamer.java b/src/java/org/apache/cassandra/dht/RangeStreamer.java
index eabb21261c..439ebc6307 100644
--- a/src/java/org/apache/cassandra/dht/RangeStreamer.java
+++ b/src/java/org/apache/cassandra/dht/RangeStreamer.java
@@ -17,7 +17,6 @@
*/
package org.apache.cassandra.dht;
-import java.net.InetAddress;
import java.util.*;
import com.google.common.annotations.VisibleForTesting;
@@ -26,8 +25,9 @@ import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.LocalStrategy;
-import org.apache.cassandra.service.ActiveRepairService;
+
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -59,10 +59,10 @@ public class RangeStreamer
/* current token ring */
private final TokenMetadata metadata;
/* address of this node */
- private final InetAddress address;
+ private final InetAddressAndPort address;
/* streaming description */
private final String description;
- private final Multimap>>> toFetch = HashMultimap.create();
+ private final Multimap>>> toFetch = HashMultimap.create();
private final Set sourceFilters = new HashSet<>();
private final StreamPlan streamPlan;
private final boolean useStrictConsistency;
@@ -74,7 +74,7 @@ public class RangeStreamer
*/
public static interface ISourceFilter
{
- public boolean shouldInclude(InetAddress endpoint);
+ public boolean shouldInclude(InetAddressAndPort endpoint);
}
/**
@@ -90,7 +90,7 @@ public class RangeStreamer
this.fd = fd;
}
- public boolean shouldInclude(InetAddress endpoint)
+ public boolean shouldInclude(InetAddressAndPort endpoint)
{
return fd.isAlive(endpoint);
}
@@ -110,7 +110,7 @@ public class RangeStreamer
this.snitch = snitch;
}
- public boolean shouldInclude(InetAddress endpoint)
+ public boolean shouldInclude(InetAddressAndPort endpoint)
{
return snitch.getDatacenter(endpoint).equals(sourceDc);
}
@@ -121,9 +121,9 @@ public class RangeStreamer
*/
public static class ExcludeLocalNodeFilter implements ISourceFilter
{
- public boolean shouldInclude(InetAddress endpoint)
+ public boolean shouldInclude(InetAddressAndPort endpoint)
{
- return !FBUtilities.getBroadcastAddress().equals(endpoint);
+ return !FBUtilities.getBroadcastAddressAndPort().equals(endpoint);
}
}
@@ -132,14 +132,14 @@ public class RangeStreamer
*/
public static class WhitelistedSourcesFilter implements ISourceFilter
{
- private final Set whitelistedSources;
+ private final Set whitelistedSources;
- public WhitelistedSourcesFilter(Set whitelistedSources)
+ public WhitelistedSourcesFilter(Set whitelistedSources)
{
this.whitelistedSources = whitelistedSources;
}
- public boolean shouldInclude(InetAddress endpoint)
+ public boolean shouldInclude(InetAddressAndPort endpoint)
{
return whitelistedSources.contains(endpoint);
}
@@ -147,7 +147,7 @@ public class RangeStreamer
public RangeStreamer(TokenMetadata metadata,
Collection tokens,
- InetAddress address,
+ InetAddressAndPort address,
StreamOperation streamOperation,
boolean useStrictConsistency,
IEndpointSnitch snitch,
@@ -186,18 +186,18 @@ public class RangeStreamer
}
boolean useStrictSource = useStrictSourcesForRanges(keyspaceName);
- Multimap, InetAddress> rangesForKeyspace = useStrictSource
+ Multimap, InetAddressAndPort> rangesForKeyspace = useStrictSource
? getAllRangesWithStrictSourcesFor(keyspaceName, ranges) : getAllRangesWithSourcesFor(keyspaceName, ranges);
- for (Map.Entry, InetAddress> entry : rangesForKeyspace.entries())
+ for (Map.Entry, InetAddressAndPort> entry : rangesForKeyspace.entries())
logger.info("{}: range {} exists on {} for keyspace {}", description, entry.getKey(), entry.getValue(), keyspaceName);
AbstractReplicationStrategy strat = Keyspace.open(keyspaceName).getReplicationStrategy();
- Multimap> rangeFetchMap = useStrictSource || strat == null || strat.getReplicationFactor() == 1
+ Multimap> rangeFetchMap = useStrictSource || strat == null || strat.getReplicationFactor() == 1
? getRangeFetchMap(rangesForKeyspace, sourceFilters, keyspaceName, useStrictConsistency)
: getOptimizedRangeFetchMap(rangesForKeyspace, sourceFilters, keyspaceName);
- for (Map.Entry>> entry : rangeFetchMap.asMap().entrySet())
+ for (Map.Entry>> entry : rangeFetchMap.asMap().entrySet())
{
if (logger.isTraceEnabled())
{
@@ -226,19 +226,19 @@ public class RangeStreamer
*
* @throws java.lang.IllegalStateException when there is no source to get data streamed
*/
- private Multimap, InetAddress> getAllRangesWithSourcesFor(String keyspaceName, Collection> desiredRanges)
+ private Multimap, InetAddressAndPort> getAllRangesWithSourcesFor(String keyspaceName, Collection> desiredRanges)
{
AbstractReplicationStrategy strat = Keyspace.open(keyspaceName).getReplicationStrategy();
- Multimap, InetAddress> rangeAddresses = strat.getRangeAddresses(metadata.cloneOnlyTokenMap());
+ Multimap, InetAddressAndPort> rangeAddresses = strat.getRangeAddresses(metadata.cloneOnlyTokenMap());
- Multimap, InetAddress> rangeSources = ArrayListMultimap.create();
+ Multimap, InetAddressAndPort> rangeSources = ArrayListMultimap.create();
for (Range desiredRange : desiredRanges)
{
for (Range range : rangeAddresses.keySet())
{
if (range.contains(desiredRange))
{
- List preferred = snitch.getSortedListByProximity(address, rangeAddresses.get(range));
+ List preferred = snitch.getSortedListByProximity(address, rangeAddresses.get(range));
rangeSources.putAll(desiredRange, preferred);
break;
}
@@ -258,30 +258,30 @@ public class RangeStreamer
*
* @throws java.lang.IllegalStateException when there is no source to get data streamed, or more than 1 source found.
*/
- private Multimap, InetAddress> getAllRangesWithStrictSourcesFor(String keyspace, Collection> desiredRanges)
+ private Multimap, InetAddressAndPort> getAllRangesWithStrictSourcesFor(String keyspace, Collection> desiredRanges)
{
assert tokens != null;
AbstractReplicationStrategy strat = Keyspace.open(keyspace).getReplicationStrategy();
// Active ranges
TokenMetadata metadataClone = metadata.cloneOnlyTokenMap();
- Multimap, InetAddress> addressRanges = strat.getRangeAddresses(metadataClone);
+ Multimap, InetAddressAndPort> addressRanges = strat.getRangeAddresses(metadataClone);
// Pending ranges
metadataClone.updateNormalTokens(tokens, address);
- Multimap, InetAddress> pendingRangeAddresses = strat.getRangeAddresses(metadataClone);
+ Multimap, InetAddressAndPort> pendingRangeAddresses = strat.getRangeAddresses(metadataClone);
// Collects the source that will have its range moved to the new node
- Multimap, InetAddress> rangeSources = ArrayListMultimap.create();
+ Multimap, InetAddressAndPort> rangeSources = ArrayListMultimap.create();
for (Range desiredRange : desiredRanges)
{
- for (Map.Entry, Collection> preEntry : addressRanges.asMap().entrySet())
+ for (Map.Entry, Collection> preEntry : addressRanges.asMap().entrySet())
{
if (preEntry.getKey().contains(desiredRange))
{
- Set oldEndpoints = Sets.newHashSet(preEntry.getValue());
- Set newEndpoints = Sets.newHashSet(pendingRangeAddresses.get(desiredRange));
+ Set oldEndpoints = Sets.newHashSet(preEntry.getValue());
+ Set newEndpoints = Sets.newHashSet(pendingRangeAddresses.get(desiredRange));
// Due to CASSANDRA-5953 we can have a higher RF then we have endpoints.
// So we need to be careful to only be strict when endpoints == RF
@@ -296,14 +296,14 @@ public class RangeStreamer
}
// Validate
- Collection