You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by aw...@apache.org on 2018/01/25 20:11:59 UTC
[02/19] cassandra git commit: Allow storage port to be configurable
per node
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/test/unit/org/apache/cassandra/repair/messages/RepairMessageSerializationsTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/repair/messages/RepairMessageSerializationsTest.java b/test/unit/org/apache/cassandra/repair/messages/RepairMessageSerializationsTest.java
index 2ef9446..e57ab94 100644
--- a/test/unit/org/apache/cassandra/repair/messages/RepairMessageSerializationsTest.java
+++ b/test/unit/org/apache/cassandra/repair/messages/RepairMessageSerializationsTest.java
@@ -19,7 +19,6 @@
package org.apache.cassandra.repair.messages;
import java.io.IOException;
-import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -44,6 +43,7 @@ import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputBufferFixed;
import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.repair.NodePair;
import org.apache.cassandra.streaming.PreviewKind;
@@ -148,9 +148,9 @@ public class RepairMessageSerializationsTest
@Test
public void syncRequestMessage() throws IOException
{
- InetAddress initiator = InetAddress.getByName("127.0.0.1");
- InetAddress src = InetAddress.getByName("127.0.0.2");
- InetAddress dst = InetAddress.getByName("127.0.0.3");
+ InetAddressAndPort initiator = InetAddressAndPort.getByName("127.0.0.1");
+ InetAddressAndPort src = InetAddressAndPort.getByName("127.0.0.2");
+ InetAddressAndPort dst = InetAddressAndPort.getByName("127.0.0.3");
SyncRequest msg = new SyncRequest(buildRepairJobDesc(), initiator, src, dst, buildTokenRanges(), PreviewKind.NONE);
serializeRoundTrip(msg, SyncRequest.serializer);
@@ -159,8 +159,8 @@ public class RepairMessageSerializationsTest
@Test
public void syncCompleteMessage() throws IOException
{
- InetAddress src = InetAddress.getByName("127.0.0.2");
- InetAddress dst = InetAddress.getByName("127.0.0.3");
+ InetAddressAndPort src = InetAddressAndPort.getByName("127.0.0.2");
+ InetAddressAndPort dst = InetAddressAndPort.getByName("127.0.0.3");
List<SessionSummary> summaries = new ArrayList<>();
summaries.add(new SessionSummary(src, dst,
Lists.newArrayList(new StreamSummary(TableId.fromUUID(UUIDGen.getTimeUUID()), 5, 100)),
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/test/unit/org/apache/cassandra/repair/messages/RepairMessageSerializerTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/repair/messages/RepairMessageSerializerTest.java b/test/unit/org/apache/cassandra/repair/messages/RepairMessageSerializerTest.java
index 9789b38..d876139 100644
--- a/test/unit/org/apache/cassandra/repair/messages/RepairMessageSerializerTest.java
+++ b/test/unit/org/apache/cassandra/repair/messages/RepairMessageSerializerTest.java
@@ -19,7 +19,6 @@
package org.apache.cassandra.repair.messages;
import java.io.IOException;
-import java.net.InetAddress;
import com.google.common.collect.Sets;
import org.junit.Assert;
@@ -27,6 +26,7 @@ import org.junit.Test;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.utils.UUIDGen;
@@ -58,10 +58,10 @@ public class RepairMessageSerializerTest
@Test
public void prepareConsistentRequest() throws Exception
{
- InetAddress coordinator = InetAddress.getByName("10.0.0.1");
- InetAddress peer1 = InetAddress.getByName("10.0.0.2");
- InetAddress peer2 = InetAddress.getByName("10.0.0.3");
- InetAddress peer3 = InetAddress.getByName("10.0.0.4");
+ InetAddressAndPort coordinator = InetAddressAndPort.getByName("10.0.0.1");
+ InetAddressAndPort peer1 = InetAddressAndPort.getByName("10.0.0.2");
+ InetAddressAndPort peer2 = InetAddressAndPort.getByName("10.0.0.3");
+ InetAddressAndPort peer3 = InetAddressAndPort.getByName("10.0.0.4");
RepairMessage expected = new PrepareConsistentRequest(UUIDGen.getTimeUUID(),
coordinator,
Sets.newHashSet(peer1, peer2, peer3));
@@ -73,7 +73,7 @@ public class RepairMessageSerializerTest
public void prepareConsistentResponse() throws Exception
{
RepairMessage expected = new PrepareConsistentResponse(UUIDGen.getTimeUUID(),
- InetAddress.getByName("10.0.0.2"),
+ InetAddressAndPort.getByName("10.0.0.2"),
true);
RepairMessage actual = serdes(expected);
Assert.assertEquals(expected, actual);
@@ -99,7 +99,7 @@ public class RepairMessageSerializerTest
public void finalizePromise() throws Exception
{
RepairMessage expected = new FinalizePromise(UUIDGen.getTimeUUID(),
- InetAddress.getByName("10.0.0.2"),
+ InetAddressAndPort.getByName("10.0.0.2"),
true);
RepairMessage actual = serdes(expected);
Assert.assertEquals(expected, actual);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java b/test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java
index cbacaec..76f915e 100644
--- a/test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java
+++ b/test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java
@@ -18,7 +18,6 @@
*/
package org.apache.cassandra.service;
-import java.net.InetAddress;
import java.util.*;
import javax.xml.crypto.Data;
@@ -43,6 +42,7 @@ import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.repair.messages.RepairOption;
import org.apache.cassandra.streaming.PreviewKind;
@@ -69,7 +69,7 @@ public class ActiveRepairServiceTest
public String cfname;
public ColumnFamilyStore store;
- public InetAddress LOCAL, REMOTE;
+ public InetAddressAndPort LOCAL, REMOTE;
private boolean initialized;
@@ -91,9 +91,9 @@ public class ActiveRepairServiceTest
SchemaLoader.startGossiper();
initialized = true;
- LOCAL = FBUtilities.getBroadcastAddress();
+ LOCAL = FBUtilities.getBroadcastAddressAndPort();
// generate a fake endpoint for which we can spoof receiving/sending trees
- REMOTE = InetAddress.getByName("127.0.0.2");
+ REMOTE = InetAddressAndPort.getByName("127.0.0.2");
}
TokenMetadata tmd = StorageService.instance.getTokenMetadata();
@@ -107,10 +107,10 @@ public class ActiveRepairServiceTest
public void testGetNeighborsPlusOne() throws Throwable
{
// generate rf+1 nodes, and ensure that all nodes are returned
- Set<InetAddress> expected = addTokens(1 + Keyspace.open(KEYSPACE5).getReplicationStrategy().getReplicationFactor());
- expected.remove(FBUtilities.getBroadcastAddress());
+ Set<InetAddressAndPort> expected = addTokens(1 + Keyspace.open(KEYSPACE5).getReplicationStrategy().getReplicationFactor());
+ expected.remove(FBUtilities.getBroadcastAddressAndPort());
Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(KEYSPACE5);
- Set<InetAddress> neighbors = new HashSet<>();
+ Set<InetAddressAndPort> neighbors = new HashSet<>();
for (Range<Token> range : ranges)
{
neighbors.addAll(ActiveRepairService.getNeighbors(KEYSPACE5, ranges, range, null, null));
@@ -126,14 +126,14 @@ public class ActiveRepairServiceTest
// generate rf*2 nodes, and ensure that only neighbors specified by the ARS are returned
addTokens(2 * Keyspace.open(KEYSPACE5).getReplicationStrategy().getReplicationFactor());
AbstractReplicationStrategy ars = Keyspace.open(KEYSPACE5).getReplicationStrategy();
- Set<InetAddress> expected = new HashSet<>();
- for (Range<Token> replicaRange : ars.getAddressRanges().get(FBUtilities.getBroadcastAddress()))
+ Set<InetAddressAndPort> expected = new HashSet<>();
+ for (Range<Token> replicaRange : ars.getAddressRanges().get(FBUtilities.getBroadcastAddressAndPort()))
{
expected.addAll(ars.getRangeAddresses(tmd.cloneOnlyTokenMap()).get(replicaRange));
}
- expected.remove(FBUtilities.getBroadcastAddress());
+ expected.remove(FBUtilities.getBroadcastAddressAndPort());
Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(KEYSPACE5);
- Set<InetAddress> neighbors = new HashSet<>();
+ Set<InetAddressAndPort> neighbors = new HashSet<>();
for (Range<Token> range : ranges)
{
neighbors.addAll(ActiveRepairService.getNeighbors(KEYSPACE5, ranges, range, null, null));
@@ -147,15 +147,15 @@ public class ActiveRepairServiceTest
TokenMetadata tmd = StorageService.instance.getTokenMetadata();
// generate rf+1 nodes, and ensure that all nodes are returned
- Set<InetAddress> expected = addTokens(1 + Keyspace.open(KEYSPACE5).getReplicationStrategy().getReplicationFactor());
- expected.remove(FBUtilities.getBroadcastAddress());
+ Set<InetAddressAndPort> expected = addTokens(1 + Keyspace.open(KEYSPACE5).getReplicationStrategy().getReplicationFactor());
+ expected.remove(FBUtilities.getBroadcastAddressAndPort());
// remove remote endpoints
TokenMetadata.Topology topology = tmd.cloneOnlyTokenMap().getTopology();
- HashSet<InetAddress> localEndpoints = Sets.newHashSet(topology.getDatacenterEndpoints().get(DatabaseDescriptor.getLocalDataCenter()));
+ HashSet<InetAddressAndPort> localEndpoints = Sets.newHashSet(topology.getDatacenterEndpoints().get(DatabaseDescriptor.getLocalDataCenter()));
expected = Sets.intersection(expected, localEndpoints);
Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(KEYSPACE5);
- Set<InetAddress> neighbors = new HashSet<>();
+ Set<InetAddressAndPort> neighbors = new HashSet<>();
for (Range<Token> range : ranges)
{
neighbors.addAll(ActiveRepairService.getNeighbors(KEYSPACE5, ranges, range, Arrays.asList(DatabaseDescriptor.getLocalDataCenter()), null));
@@ -171,19 +171,19 @@ public class ActiveRepairServiceTest
// generate rf*2 nodes, and ensure that only neighbors specified by the ARS are returned
addTokens(2 * Keyspace.open(KEYSPACE5).getReplicationStrategy().getReplicationFactor());
AbstractReplicationStrategy ars = Keyspace.open(KEYSPACE5).getReplicationStrategy();
- Set<InetAddress> expected = new HashSet<>();
- for (Range<Token> replicaRange : ars.getAddressRanges().get(FBUtilities.getBroadcastAddress()))
+ Set<InetAddressAndPort> expected = new HashSet<>();
+ for (Range<Token> replicaRange : ars.getAddressRanges().get(FBUtilities.getBroadcastAddressAndPort()))
{
expected.addAll(ars.getRangeAddresses(tmd.cloneOnlyTokenMap()).get(replicaRange));
}
- expected.remove(FBUtilities.getBroadcastAddress());
+ expected.remove(FBUtilities.getBroadcastAddressAndPort());
// remove remote endpoints
TokenMetadata.Topology topology = tmd.cloneOnlyTokenMap().getTopology();
- HashSet<InetAddress> localEndpoints = Sets.newHashSet(topology.getDatacenterEndpoints().get(DatabaseDescriptor.getLocalDataCenter()));
+ HashSet<InetAddressAndPort> localEndpoints = Sets.newHashSet(topology.getDatacenterEndpoints().get(DatabaseDescriptor.getLocalDataCenter()));
expected = Sets.intersection(expected, localEndpoints);
Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(KEYSPACE5);
- Set<InetAddress> neighbors = new HashSet<>();
+ Set<InetAddressAndPort> neighbors = new HashSet<>();
for (Range<Token> range : ranges)
{
neighbors.addAll(ActiveRepairService.getNeighbors(KEYSPACE5, ranges, range, Arrays.asList(DatabaseDescriptor.getLocalDataCenter()), null));
@@ -199,14 +199,14 @@ public class ActiveRepairServiceTest
// generate rf*2 nodes, and ensure that only neighbors specified by the hosts are returned
addTokens(2 * Keyspace.open(KEYSPACE5).getReplicationStrategy().getReplicationFactor());
AbstractReplicationStrategy ars = Keyspace.open(KEYSPACE5).getReplicationStrategy();
- List<InetAddress> expected = new ArrayList<>();
- for (Range<Token> replicaRange : ars.getAddressRanges().get(FBUtilities.getBroadcastAddress()))
+ List<InetAddressAndPort> expected = new ArrayList<>();
+ for (Range<Token> replicaRange : ars.getAddressRanges().get(FBUtilities.getBroadcastAddressAndPort()))
{
expected.addAll(ars.getRangeAddresses(tmd.cloneOnlyTokenMap()).get(replicaRange));
}
- expected.remove(FBUtilities.getBroadcastAddress());
- Collection<String> hosts = Arrays.asList(FBUtilities.getBroadcastAddress().getCanonicalHostName(),expected.get(0).getCanonicalHostName());
+ expected.remove(FBUtilities.getBroadcastAddressAndPort());
+ Collection<String> hosts = Arrays.asList(FBUtilities.getBroadcastAddressAndPort().toString(),expected.get(0).toString());
Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(KEYSPACE5);
assertEquals(expected.get(0), ActiveRepairService.getNeighbors(KEYSPACE5, ranges,
@@ -245,13 +245,13 @@ public class ActiveRepairServiceTest
}
- Set<InetAddress> addTokens(int max) throws Throwable
+ Set<InetAddressAndPort> addTokens(int max) throws Throwable
{
TokenMetadata tmd = StorageService.instance.getTokenMetadata();
- Set<InetAddress> endpoints = new HashSet<>();
+ Set<InetAddressAndPort> endpoints = new HashSet<>();
for (int i = 1; i <= max; i++)
{
- InetAddress endpoint = InetAddress.getByName("127.0.0." + i);
+ InetAddressAndPort endpoint = InetAddressAndPort.getByName("127.0.0." + i);
tmd.updateNormalToken(tmd.partitioner.getRandomToken(), endpoint);
endpoints.add(endpoint);
}
@@ -264,14 +264,14 @@ public class ActiveRepairServiceTest
ColumnFamilyStore store = prepareColumnFamilyStore();
UUID prsId = UUID.randomUUID();
Set<SSTableReader> original = Sets.newHashSet(store.select(View.select(SSTableSet.CANONICAL, (s) -> !s.isRepaired())).sstables);
- ActiveRepairService.instance.registerParentRepairSession(prsId, FBUtilities.getBroadcastAddress(), Collections.singletonList(store),
+ ActiveRepairService.instance.registerParentRepairSession(prsId, FBUtilities.getBroadcastAddressAndPort(), Collections.singletonList(store),
Collections.singleton(new Range<>(store.getPartitioner().getMinimumToken(),
store.getPartitioner().getMinimumToken())),
true, System.currentTimeMillis(), true, PreviewKind.NONE);
ActiveRepairService.instance.getParentRepairSession(prsId).maybeSnapshot(store.metadata.id, prsId);
UUID prsId2 = UUID.randomUUID();
- ActiveRepairService.instance.registerParentRepairSession(prsId2, FBUtilities.getBroadcastAddress(),
+ ActiveRepairService.instance.registerParentRepairSession(prsId2, FBUtilities.getBroadcastAddressAndPort(),
Collections.singletonList(store),
Collections.singleton(new Range<>(store.getPartitioner().getMinimumToken(),
store.getPartitioner().getMinimumToken())),
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/test/unit/org/apache/cassandra/service/DataResolverTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/service/DataResolverTest.java b/test/unit/org/apache/cassandra/service/DataResolverTest.java
index ec25fef..f406582 100644
--- a/test/unit/org/apache/cassandra/service/DataResolverTest.java
+++ b/test/unit/org/apache/cassandra/service/DataResolverTest.java
@@ -18,7 +18,6 @@
*/
package org.apache.cassandra.service;
-import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.*;
@@ -29,6 +28,7 @@ import org.junit.*;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.Util;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.config.DatabaseDescriptor;
@@ -150,11 +150,11 @@ public class DataResolverTest
public void testResolveNewerSingleRow() throws UnknownHostException
{
DataResolver resolver = new DataResolver(ks, command, ConsistencyLevel.ALL, 2, System.nanoTime());
- InetAddress peer1 = peer();
+ InetAddressAndPort peer1 = peer();
resolver.preprocess(readResponseMessage(peer1, iter(new RowUpdateBuilder(cfm, nowInSec, 0L, dk).clustering("1")
.add("c1", "v1")
.buildUpdate())));
- InetAddress peer2 = peer();
+ InetAddressAndPort peer2 = peer();
resolver.preprocess(readResponseMessage(peer2, iter(new RowUpdateBuilder(cfm, nowInSec, 1L, dk).clustering("1")
.add("c1", "v2")
.buildUpdate())));
@@ -182,12 +182,12 @@ public class DataResolverTest
public void testResolveDisjointSingleRow()
{
DataResolver resolver = new DataResolver(ks, command, ConsistencyLevel.ALL, 2, System.nanoTime());
- InetAddress peer1 = peer();
+ InetAddressAndPort peer1 = peer();
resolver.preprocess(readResponseMessage(peer1, iter(new RowUpdateBuilder(cfm, nowInSec, 0L, dk).clustering("1")
.add("c1", "v1")
.buildUpdate())));
- InetAddress peer2 = peer();
+ InetAddressAndPort peer2 = peer();
resolver.preprocess(readResponseMessage(peer2, iter(new RowUpdateBuilder(cfm, nowInSec, 1L, dk).clustering("1")
.add("c2", "v2")
.buildUpdate())));
@@ -220,11 +220,11 @@ public class DataResolverTest
{
DataResolver resolver = new DataResolver(ks, command, ConsistencyLevel.ALL, 2, System.nanoTime());
- InetAddress peer1 = peer();
+ InetAddressAndPort peer1 = peer();
resolver.preprocess(readResponseMessage(peer1, iter(new RowUpdateBuilder(cfm, nowInSec, 0L, dk).clustering("1")
.add("c1", "v1")
.buildUpdate())));
- InetAddress peer2 = peer();
+ InetAddressAndPort peer2 = peer();
resolver.preprocess(readResponseMessage(peer2, iter(new RowUpdateBuilder(cfm, nowInSec, 1L, dk).clustering("2")
.add("c2", "v2")
.buildUpdate())));
@@ -274,25 +274,25 @@ public class DataResolverTest
.addRangeTombstone(tombstone2)
.buildUpdate();
- InetAddress peer1 = peer();
+ InetAddressAndPort peer1 = peer();
UnfilteredPartitionIterator iter1 = iter(new RowUpdateBuilder(cfm, nowInSec, 1L, dk).addRangeTombstone(tombstone1)
.addRangeTombstone(tombstone2)
.buildUpdate());
resolver.preprocess(readResponseMessage(peer1, iter1));
// not covered by any range tombstone
- InetAddress peer2 = peer();
+ InetAddressAndPort peer2 = peer();
UnfilteredPartitionIterator iter2 = iter(new RowUpdateBuilder(cfm, nowInSec, 0L, dk).clustering("0")
.add("c1", "v0")
.buildUpdate());
resolver.preprocess(readResponseMessage(peer2, iter2));
// covered by a range tombstone
- InetAddress peer3 = peer();
+ InetAddressAndPort peer3 = peer();
UnfilteredPartitionIterator iter3 = iter(new RowUpdateBuilder(cfm, nowInSec, 0L, dk).clustering("10")
.add("c2", "v1")
.buildUpdate());
resolver.preprocess(readResponseMessage(peer3, iter3));
// range covered by rt, but newer
- InetAddress peer4 = peer();
+ InetAddressAndPort peer4 = peer();
UnfilteredPartitionIterator iter4 = iter(new RowUpdateBuilder(cfm, nowInSec, 2L, dk).clustering("3")
.add("one", "A")
.buildUpdate());
@@ -348,11 +348,11 @@ public class DataResolverTest
public void testResolveWithOneEmpty()
{
DataResolver resolver = new DataResolver(ks, command, ConsistencyLevel.ALL, 2, System.nanoTime());
- InetAddress peer1 = peer();
+ InetAddressAndPort peer1 = peer();
resolver.preprocess(readResponseMessage(peer1, iter(new RowUpdateBuilder(cfm, nowInSec, 1L, dk).clustering("1")
.add("c2", "v2")
.buildUpdate())));
- InetAddress peer2 = peer();
+ InetAddressAndPort peer2 = peer();
resolver.preprocess(readResponseMessage(peer2, EmptyIterators.unfilteredPartition(cfm)));
try(PartitionIterator data = resolver.resolve())
@@ -395,11 +395,11 @@ public class DataResolverTest
{
DataResolver resolver = new DataResolver(ks, command, ConsistencyLevel.ALL, 2, System.nanoTime());
// one response with columns timestamped before a delete in another response
- InetAddress peer1 = peer();
+ InetAddressAndPort peer1 = peer();
resolver.preprocess(readResponseMessage(peer1, iter(new RowUpdateBuilder(cfm, nowInSec, 0L, dk).clustering("1")
.add("one", "A")
.buildUpdate())));
- InetAddress peer2 = peer();
+ InetAddressAndPort peer2 = peer();
resolver.preprocess(readResponseMessage(peer2, fullPartitionDelete(cfm, dk, 1, nowInSec)));
try (PartitionIterator data = resolver.resolve())
@@ -421,20 +421,20 @@ public class DataResolverTest
{
DataResolver resolver = new DataResolver(ks, command, ConsistencyLevel.ALL, 4, System.nanoTime());
// deletes and columns with interleaved timestamp, with out of order return sequence
- InetAddress peer1 = peer();
+ InetAddressAndPort peer1 = peer();
resolver.preprocess(readResponseMessage(peer1, fullPartitionDelete(cfm, dk, 0, nowInSec)));
// these columns created after the previous deletion
- InetAddress peer2 = peer();
+ InetAddressAndPort peer2 = peer();
resolver.preprocess(readResponseMessage(peer2, iter(new RowUpdateBuilder(cfm, nowInSec, 1L, dk).clustering("1")
.add("one", "A")
.add("two", "A")
.buildUpdate())));
//this column created after the next delete
- InetAddress peer3 = peer();
+ InetAddressAndPort peer3 = peer();
resolver.preprocess(readResponseMessage(peer3, iter(new RowUpdateBuilder(cfm, nowInSec, 3L, dk).clustering("1")
.add("two", "B")
.buildUpdate())));
- InetAddress peer4 = peer();
+ InetAddressAndPort peer4 = peer();
resolver.preprocess(readResponseMessage(peer4, fullPartitionDelete(cfm, dk, 2, nowInSec)));
try(PartitionIterator data = resolver.resolve())
@@ -505,8 +505,8 @@ public class DataResolverTest
private void resolveRangeTombstonesOnBoundary(long timestamp1, long timestamp2)
{
DataResolver resolver = new DataResolver(ks, command, ConsistencyLevel.ALL, 2, System.nanoTime());
- InetAddress peer1 = peer();
- InetAddress peer2 = peer();
+ InetAddressAndPort peer1 = peer();
+ InetAddressAndPort peer2 = peer();
// 1st "stream"
RangeTombstone one_two = tombstone("1", true , "2", false, timestamp1, nowInSec);
@@ -579,8 +579,8 @@ public class DataResolverTest
private void testRepairRangeTombstoneBoundary(int timestamp1, int timestamp2, int timestamp3) throws UnknownHostException
{
DataResolver resolver = new DataResolver(ks, command, ConsistencyLevel.ALL, 2, System.nanoTime());
- InetAddress peer1 = peer();
- InetAddress peer2 = peer();
+ InetAddressAndPort peer1 = peer();
+ InetAddressAndPort peer2 = peer();
// 1st "stream"
RangeTombstone one_nine = tombstone("0", true , "9", true, timestamp1, nowInSec);
@@ -631,8 +631,8 @@ public class DataResolverTest
public void testRepairRangeTombstoneWithPartitionDeletion()
{
DataResolver resolver = new DataResolver(ks, command, ConsistencyLevel.ALL, 2, System.nanoTime());
- InetAddress peer1 = peer();
- InetAddress peer2 = peer();
+ InetAddressAndPort peer1 = peer();
+ InetAddressAndPort peer2 = peer();
// 1st "stream": just a partition deletion
UnfilteredPartitionIterator iter1 = iter(PartitionUpdate.fullPartitionDelete(cfm, dk, 10, nowInSec));
@@ -670,8 +670,8 @@ public class DataResolverTest
public void testRepairRangeTombstoneWithPartitionDeletion2()
{
DataResolver resolver = new DataResolver(ks, command, ConsistencyLevel.ALL, 2, System.nanoTime());
- InetAddress peer1 = peer();
- InetAddress peer2 = peer();
+ InetAddressAndPort peer1 = peer();
+ InetAddressAndPort peer2 = peer();
// 1st "stream": a partition deletion and a range tombstone
RangeTombstone rt1 = tombstone("0", true , "9", true, 11, nowInSec);
@@ -761,7 +761,7 @@ public class DataResolverTest
builder.addComplexDeletion(m, new DeletionTime(ts[0] - 1, nowInSec));
builder.addCell(mapCell(0, 0, ts[0]));
- InetAddress peer1 = peer();
+ InetAddressAndPort peer1 = peer();
resolver.preprocess(readResponseMessage(peer1, iter(PartitionUpdate.singleRowUpdate(cfm2, dk, builder.build())), cmd));
builder.newRow(Clustering.EMPTY);
@@ -770,7 +770,7 @@ public class DataResolverTest
Cell expectedCell = mapCell(1, 1, ts[1]);
builder.addCell(expectedCell);
- InetAddress peer2 = peer();
+ InetAddressAndPort peer2 = peer();
resolver.preprocess(readResponseMessage(peer2, iter(PartitionUpdate.singleRowUpdate(cfm2, dk, builder.build())), cmd));
try(PartitionIterator data = resolver.resolve())
@@ -814,14 +814,14 @@ public class DataResolverTest
builder.addComplexDeletion(m, new DeletionTime(ts[0] - 1, nowInSec));
builder.addCell(mapCell(0, 0, ts[0]));
- InetAddress peer1 = peer();
+ InetAddressAndPort peer1 = peer();
resolver.preprocess(readResponseMessage(peer1, iter(PartitionUpdate.singleRowUpdate(cfm2, dk, builder.build())), cmd));
builder.newRow(Clustering.EMPTY);
DeletionTime expectedCmplxDelete = new DeletionTime(ts[1] - 1, nowInSec);
builder.addComplexDeletion(m, expectedCmplxDelete);
- InetAddress peer2 = peer();
+ InetAddressAndPort peer2 = peer();
resolver.preprocess(readResponseMessage(peer2, iter(PartitionUpdate.singleRowUpdate(cfm2, dk, builder.build())), cmd));
try(PartitionIterator data = resolver.resolve())
@@ -862,10 +862,10 @@ public class DataResolverTest
builder.addCell(expectedCell);
// empty map column
- InetAddress peer1 = peer();
+ InetAddressAndPort peer1 = peer();
resolver.preprocess(readResponseMessage(peer1, iter(PartitionUpdate.singleRowUpdate(cfm2, dk, builder.build())), cmd));
- InetAddress peer2 = peer();
+ InetAddressAndPort peer2 = peer();
resolver.preprocess(readResponseMessage(peer2, iter(PartitionUpdate.emptyUpdate(cfm2, dk))));
try(PartitionIterator data = resolver.resolve())
@@ -908,7 +908,7 @@ public class DataResolverTest
builder.newRow(Clustering.EMPTY);
builder.addComplexDeletion(m, new DeletionTime(ts[0] - 1, nowInSec));
- InetAddress peer1 = peer();
+ InetAddressAndPort peer1 = peer();
resolver.preprocess(readResponseMessage(peer1, iter(PartitionUpdate.singleRowUpdate(cfm2, dk, builder.build())), cmd));
// newer, overwritten map column
@@ -918,7 +918,7 @@ public class DataResolverTest
Cell expectedCell = mapCell(1, 1, ts[1]);
builder.addCell(expectedCell);
- InetAddress peer2 = peer();
+ InetAddressAndPort peer2 = peer();
resolver.preprocess(readResponseMessage(peer2, iter(PartitionUpdate.singleRowUpdate(cfm2, dk, builder.build())), cmd));
try(PartitionIterator data = resolver.resolve())
@@ -945,11 +945,11 @@ public class DataResolverTest
Assert.assertNull(messageRecorder.sent.get(peer2));
}
- private InetAddress peer()
+ private InetAddressAndPort peer()
{
try
{
- return InetAddress.getByAddress(new byte[]{ 127, 0, 0, (byte) addressSuffix++ });
+ return InetAddressAndPort.getByAddress(new byte[]{ 127, 0, 0, (byte) addressSuffix++ });
}
catch (UnknownHostException e)
{
@@ -957,7 +957,7 @@ public class DataResolverTest
}
}
- private MessageOut<Mutation> getSentMessage(InetAddress target)
+ private MessageOut<Mutation> getSentMessage(InetAddressAndPort target)
{
MessageOut<Mutation> message = messageRecorder.sent.get(target);
assertNotNull(String.format("No repair message was sent to %s", target), message);
@@ -1018,12 +1018,12 @@ public class DataResolverTest
}
- public MessageIn<ReadResponse> readResponseMessage(InetAddress from, UnfilteredPartitionIterator partitionIterator)
+ public MessageIn<ReadResponse> readResponseMessage(InetAddressAndPort from, UnfilteredPartitionIterator partitionIterator)
{
return readResponseMessage(from, partitionIterator, command);
}
- public MessageIn<ReadResponse> readResponseMessage(InetAddress from, UnfilteredPartitionIterator partitionIterator, ReadCommand cmd)
+ public MessageIn<ReadResponse> readResponseMessage(InetAddressAndPort from, UnfilteredPartitionIterator partitionIterator, ReadCommand cmd)
{
return MessageIn.create(from,
ReadResponse.createRemoteDataResponse(partitionIterator, cmd),
@@ -1080,8 +1080,8 @@ public class DataResolverTest
private static class MessageRecorder implements IMessageSink
{
- Map<InetAddress, MessageOut> sent = new HashMap<>();
- public boolean allowOutgoingMessage(MessageOut message, int id, InetAddress to)
+ Map<InetAddressAndPort, MessageOut> sent = new HashMap<>();
+ public boolean allowOutgoingMessage(MessageOut message, int id, InetAddressAndPort to)
{
sent.put(to, message);
return false;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/test/unit/org/apache/cassandra/service/LeaveAndBootstrapTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/service/LeaveAndBootstrapTest.java b/test/unit/org/apache/cassandra/service/LeaveAndBootstrapTest.java
index aaa2594..8ddc4f0 100644
--- a/test/unit/org/apache/cassandra/service/LeaveAndBootstrapTest.java
+++ b/test/unit/org/apache/cassandra/service/LeaveAndBootstrapTest.java
@@ -19,7 +19,6 @@
package org.apache.cassandra.service;
-import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;
@@ -34,6 +33,7 @@ import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.Util;
import org.apache.cassandra.Util.PartitionerSwitcher;
import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.dht.IPartitioner;
@@ -92,17 +92,17 @@ public class LeaveAndBootstrapTest
ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>();
- List<InetAddress> hosts = new ArrayList<InetAddress>();
+ List<InetAddressAndPort> hosts = new ArrayList<>();
List<UUID> hostIds = new ArrayList<UUID>();
Util.createInitialRing(ss, partitioner, endpointTokens, keyTokens, hosts, hostIds, RING_SIZE);
- Map<Token, List<InetAddress>> expectedEndpoints = new HashMap<Token, List<InetAddress>>();
+ Map<Token, List<InetAddressAndPort>> expectedEndpoints = new HashMap<Token, List<InetAddressAndPort>>();
for (String keyspaceName : Schema.instance.getNonLocalStrategyKeyspaces())
{
for (Token token : keyTokens)
{
- List<InetAddress> endpoints = new ArrayList<InetAddress>();
+ List<InetAddressAndPort> endpoints = new ArrayList<>();
Iterator<Token> tokenIter = TokenMetadata.ringIterator(tmd.sortedTokens(), token, false);
while (tokenIter.hasNext())
{
@@ -114,6 +114,9 @@ public class LeaveAndBootstrapTest
// Third node leaves
ss.onChange(hosts.get(LEAVING_NODE),
+ ApplicationState.STATUS_WITH_PORT,
+ valueFactory.leaving(Collections.singleton(endpointTokens.get(LEAVING_NODE))));
+ ss.onChange(hosts.get(LEAVING_NODE),
ApplicationState.STATUS,
valueFactory.leaving(Collections.singleton(endpointTokens.get(LEAVING_NODE))));
assertTrue(tmd.isLeaving(hosts.get(LEAVING_NODE)));
@@ -129,8 +132,8 @@ public class LeaveAndBootstrapTest
{
int replicationFactor = strategy.getReplicationFactor();
- HashSet<InetAddress> actual = new HashSet<InetAddress>(tmd.getWriteEndpoints(token, keyspaceName, strategy.calculateNaturalEndpoints(token, tmd.cloneOnlyTokenMap())));
- HashSet<InetAddress> expected = new HashSet<InetAddress>();
+ HashSet<InetAddressAndPort> actual = new HashSet<>(tmd.getWriteEndpoints(token, keyspaceName, strategy.calculateNaturalEndpoints(token, tmd.cloneOnlyTokenMap())));
+ HashSet<InetAddressAndPort> expected = new HashSet<>();
for (int i = 0; i < replicationFactor; i++)
{
@@ -163,34 +166,39 @@ public class LeaveAndBootstrapTest
ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>();
- List<InetAddress> hosts = new ArrayList<InetAddress>();
+ List<InetAddressAndPort> hosts = new ArrayList<>();
List<UUID> hostIds = new ArrayList<UUID>();
// create a ring or 10 nodes
Util.createInitialRing(ss, partitioner, endpointTokens, keyTokens, hosts, hostIds, RING_SIZE);
// nodes 6, 8 and 9 leave
- final int[] LEAVING = new int[] {6, 8, 9};
+ final int[] LEAVING = new int[]{ 6, 8, 9 };
for (int leaving : LEAVING)
+ {
+ ss.onChange(hosts.get(leaving),
+ ApplicationState.STATUS_WITH_PORT,
+ valueFactory.leaving(Collections.singleton(endpointTokens.get(leaving))));
ss.onChange(hosts.get(leaving),
ApplicationState.STATUS,
valueFactory.leaving(Collections.singleton(endpointTokens.get(leaving))));
+ }
// boot two new nodes with keyTokens.get(5) and keyTokens.get(7)
- InetAddress boot1 = InetAddress.getByName("127.0.1.1");
+ InetAddressAndPort boot1 = InetAddressAndPort.getByName("127.0.1.1");
Gossiper.instance.initializeNodeUnsafe(boot1, UUID.randomUUID(), 1);
Gossiper.instance.injectApplicationState(boot1, ApplicationState.TOKENS, valueFactory.tokens(Collections.singleton(keyTokens.get(5))));
ss.onChange(boot1,
ApplicationState.STATUS,
valueFactory.bootstrapping(Collections.<Token>singleton(keyTokens.get(5))));
- InetAddress boot2 = InetAddress.getByName("127.0.1.2");
+ InetAddressAndPort boot2 = InetAddressAndPort.getByName("127.0.1.2");
Gossiper.instance.initializeNodeUnsafe(boot2, UUID.randomUUID(), 1);
Gossiper.instance.injectApplicationState(boot2, ApplicationState.TOKENS, valueFactory.tokens(Collections.singleton(keyTokens.get(7))));
ss.onChange(boot2,
ApplicationState.STATUS,
valueFactory.bootstrapping(Collections.<Token>singleton(keyTokens.get(7))));
- Collection<InetAddress> endpoints = null;
+ Collection<InetAddressAndPort> endpoints = null;
/* don't require test update every time a new keyspace is added to test/conf/cassandra.yaml */
Map<String, AbstractReplicationStrategy> keyspaceStrategyMap = new HashMap<String, AbstractReplicationStrategy>();
@@ -200,8 +208,8 @@ public class LeaveAndBootstrapTest
}
// pre-calculate the results.
- Map<String, Multimap<Token, InetAddress>> expectedEndpoints = new HashMap<String, Multimap<Token, InetAddress>>();
- expectedEndpoints.put(KEYSPACE1, HashMultimap.<Token, InetAddress>create());
+ Map<String, Multimap<Token, InetAddressAndPort>> expectedEndpoints = new HashMap<String, Multimap<Token, InetAddressAndPort>>();
+ expectedEndpoints.put(KEYSPACE1, HashMultimap.<Token, InetAddressAndPort>create());
expectedEndpoints.get(KEYSPACE1).putAll(new BigIntegerToken("5"), makeAddrs("127.0.0.2"));
expectedEndpoints.get(KEYSPACE1).putAll(new BigIntegerToken("15"), makeAddrs("127.0.0.3"));
expectedEndpoints.get(KEYSPACE1).putAll(new BigIntegerToken("25"), makeAddrs("127.0.0.4"));
@@ -212,7 +220,7 @@ public class LeaveAndBootstrapTest
expectedEndpoints.get(KEYSPACE1).putAll(new BigIntegerToken("75"), makeAddrs("127.0.0.9", "127.0.1.2", "127.0.0.1"));
expectedEndpoints.get(KEYSPACE1).putAll(new BigIntegerToken("85"), makeAddrs("127.0.0.10", "127.0.0.1"));
expectedEndpoints.get(KEYSPACE1).putAll(new BigIntegerToken("95"), makeAddrs("127.0.0.1"));
- expectedEndpoints.put(KEYSPACE2, HashMultimap.<Token, InetAddress>create());
+ expectedEndpoints.put(KEYSPACE2, HashMultimap.<Token, InetAddressAndPort>create());
expectedEndpoints.get(KEYSPACE2).putAll(new BigIntegerToken("5"), makeAddrs("127.0.0.2"));
expectedEndpoints.get(KEYSPACE2).putAll(new BigIntegerToken("15"), makeAddrs("127.0.0.3"));
expectedEndpoints.get(KEYSPACE2).putAll(new BigIntegerToken("25"), makeAddrs("127.0.0.4"));
@@ -223,7 +231,7 @@ public class LeaveAndBootstrapTest
expectedEndpoints.get(KEYSPACE2).putAll(new BigIntegerToken("75"), makeAddrs("127.0.0.9", "127.0.1.2", "127.0.0.1"));
expectedEndpoints.get(KEYSPACE2).putAll(new BigIntegerToken("85"), makeAddrs("127.0.0.10", "127.0.0.1"));
expectedEndpoints.get(KEYSPACE2).putAll(new BigIntegerToken("95"), makeAddrs("127.0.0.1"));
- expectedEndpoints.put(KEYSPACE3, HashMultimap.<Token, InetAddress>create());
+ expectedEndpoints.put(KEYSPACE3, HashMultimap.<Token, InetAddressAndPort>create());
expectedEndpoints.get(KEYSPACE3).putAll(new BigIntegerToken("5"), makeAddrs("127.0.0.2", "127.0.0.3", "127.0.0.4", "127.0.0.5", "127.0.0.6"));
expectedEndpoints.get(KEYSPACE3).putAll(new BigIntegerToken("15"), makeAddrs("127.0.0.3", "127.0.0.4", "127.0.0.5", "127.0.0.6", "127.0.0.7", "127.0.1.1", "127.0.0.8"));
expectedEndpoints.get(KEYSPACE3).putAll(new BigIntegerToken("25"), makeAddrs("127.0.0.4", "127.0.0.5", "127.0.0.6", "127.0.0.7", "127.0.0.8", "127.0.1.2", "127.0.0.1", "127.0.1.1"));
@@ -234,7 +242,7 @@ public class LeaveAndBootstrapTest
expectedEndpoints.get(KEYSPACE3).putAll(new BigIntegerToken("75"), makeAddrs("127.0.0.9", "127.0.0.10", "127.0.0.1", "127.0.0.2", "127.0.0.3", "127.0.1.2", "127.0.0.4", "127.0.0.5"));
expectedEndpoints.get(KEYSPACE3).putAll(new BigIntegerToken("85"), makeAddrs("127.0.0.10", "127.0.0.1", "127.0.0.2", "127.0.0.3", "127.0.0.4", "127.0.0.5"));
expectedEndpoints.get(KEYSPACE3).putAll(new BigIntegerToken("95"), makeAddrs("127.0.0.1", "127.0.0.2", "127.0.0.3", "127.0.0.4", "127.0.0.5"));
- expectedEndpoints.put(KEYSPACE4, HashMultimap.<Token, InetAddress>create());
+ expectedEndpoints.put(KEYSPACE4, HashMultimap.<Token, InetAddressAndPort>create());
expectedEndpoints.get(KEYSPACE4).putAll(new BigIntegerToken("5"), makeAddrs("127.0.0.2", "127.0.0.3", "127.0.0.4"));
expectedEndpoints.get(KEYSPACE4).putAll(new BigIntegerToken("15"), makeAddrs("127.0.0.3", "127.0.0.4", "127.0.0.5"));
expectedEndpoints.get(KEYSPACE4).putAll(new BigIntegerToken("25"), makeAddrs("127.0.0.4", "127.0.0.5", "127.0.0.6"));
@@ -463,7 +471,7 @@ public class LeaveAndBootstrapTest
ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>();
- List<InetAddress> hosts = new ArrayList<InetAddress>();
+ List<InetAddressAndPort> hosts = new ArrayList<>();
List<UUID> hostIds = new ArrayList<UUID>();
// create a ring or 5 nodes
@@ -540,7 +548,7 @@ public class LeaveAndBootstrapTest
ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>();
- List<InetAddress> hosts = new ArrayList<InetAddress>();
+ List<InetAddressAndPort> hosts = new ArrayList<>();
List<UUID> hostIds = new ArrayList<UUID>();
// create a ring or 5 nodes
@@ -582,7 +590,7 @@ public class LeaveAndBootstrapTest
ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>();
- List<InetAddress> hosts = new ArrayList<InetAddress>();
+ List<InetAddressAndPort> hosts = new ArrayList<>();
List<UUID> hostIds = new ArrayList<UUID>();
// create a ring or 5 nodes
@@ -632,7 +640,7 @@ public class LeaveAndBootstrapTest
ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>();
- List<InetAddress> hosts = new ArrayList<InetAddress>();
+ List<InetAddressAndPort> hosts = new ArrayList<>();
List<UUID> hostIds = new ArrayList<UUID>();
// create a ring of 6 nodes
@@ -673,10 +681,10 @@ public class LeaveAndBootstrapTest
// create a ring of 2 nodes
ArrayList<Token> endpointTokens = new ArrayList<>();
- List<InetAddress> hosts = new ArrayList<>();
+ List<InetAddressAndPort> hosts = new ArrayList<>();
Util.createInitialRing(ss, partitioner, endpointTokens, new ArrayList<Token>(), hosts, new ArrayList<UUID>(), 2);
- InetAddress toRemove = hosts.get(1);
+ InetAddressAndPort toRemove = hosts.get(1);
SystemKeyspace.updatePeerInfo(toRemove, "data_center", "dc42");
SystemKeyspace.updatePeerInfo(toRemove, "rack", "rack42");
assertEquals("rack42", SystemKeyspace.loadDcRackInfo().get(toRemove).get("rack"));
@@ -698,18 +706,19 @@ public class LeaveAndBootstrapTest
// create a ring of 1 node
StorageService ss = StorageService.instance;
VersionedValue.VersionedValueFactory valueFactory = new VersionedValue.VersionedValueFactory(partitioner);
- Util.createInitialRing(ss, partitioner, new ArrayList<Token>(), new ArrayList<Token>(), new ArrayList<InetAddress>(), new ArrayList<UUID>(), 1);
+ Util.createInitialRing(ss, partitioner, new ArrayList<Token>(), new ArrayList<Token>(), new ArrayList<InetAddressAndPort>(), new ArrayList<UUID>(), 1);
// make a REMOVING state change on a non-member endpoint; without the CASSANDRA-6564 fix, this
// would result in an ArrayIndexOutOfBoundsException
- ss.onChange(InetAddress.getByName("192.168.1.42"), ApplicationState.STATUS, valueFactory.removingNonlocal(UUID.randomUUID()));
+ ss.onChange(InetAddressAndPort.getByName("192.168.1.42"), ApplicationState.STATUS_WITH_PORT, valueFactory.removingNonlocal(UUID.randomUUID()));
+ ss.onChange(InetAddressAndPort.getByName("192.168.1.42"), ApplicationState.STATUS, valueFactory.removingNonlocal(UUID.randomUUID()));
}
- private static Collection<InetAddress> makeAddrs(String... hosts) throws UnknownHostException
+ private static Collection<InetAddressAndPort> makeAddrs(String... hosts) throws UnknownHostException
{
- ArrayList<InetAddress> addrs = new ArrayList<InetAddress>(hosts.length);
+ ArrayList<InetAddressAndPort> addrs = new ArrayList<>(hosts.length);
for (String host : hosts)
- addrs.add(InetAddress.getByName(host));
+ addrs.add(InetAddressAndPort.getByName(host));
return addrs;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/test/unit/org/apache/cassandra/service/MoveTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/service/MoveTest.java b/test/unit/org/apache/cassandra/service/MoveTest.java
index df4e294..7321fba 100644
--- a/test/unit/org/apache/cassandra/service/MoveTest.java
+++ b/test/unit/org/apache/cassandra/service/MoveTest.java
@@ -19,7 +19,6 @@
package org.apache.cassandra.service;
-import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;
@@ -30,6 +29,7 @@ import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.schema.MigrationManager;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.config.DatabaseDescriptor;
@@ -115,7 +115,7 @@ public class MoveTest
//Odd IPs are in DC1 and Even are in DC2. Endpoints upto .14 will have unique racks and
// then will be same for a set of three.
@Override
- public String getRack(InetAddress endpoint)
+ public String getRack(InetAddressAndPort endpoint)
{
int ipLastPart = getIPLastPart(endpoint);
if (ipLastPart <= 14)
@@ -125,7 +125,7 @@ public class MoveTest
}
@Override
- public String getDatacenter(InetAddress endpoint)
+ public String getDatacenter(InetAddressAndPort endpoint)
{
if (getIPLastPart(endpoint) % 2 == 0)
return "DC2";
@@ -133,19 +133,19 @@ public class MoveTest
return "DC1";
}
- private int getIPLastPart(InetAddress endpoint)
+ private int getIPLastPart(InetAddressAndPort endpoint)
{
String str = endpoint.toString();
int index = str.lastIndexOf(".");
- return Integer.parseInt(str.substring(index + 1).trim());
+ return Integer.parseInt(str.substring(index + 1).trim().split(":")[0]);
}
});
final TokenMetadata tmd = StorageService.instance.getTokenMetadata();
tmd.clearUnsafe();
- tmd.updateHostId(UUID.randomUUID(), InetAddress.getByName("127.0.0.1"));
- tmd.updateHostId(UUID.randomUUID(), InetAddress.getByName("127.0.0.2"));
+ tmd.updateHostId(UUID.randomUUID(), InetAddressAndPort.getByName("127.0.0.1"));
+ tmd.updateHostId(UUID.randomUUID(), InetAddressAndPort.getByName("127.0.0.2"));
KeyspaceMetadata keyspace = KeyspaceMetadata.create(keyspaceName,
KeyspaceParams.nts(configOptions(replicas)),
@@ -181,7 +181,7 @@ public class MoveTest
VersionedValue.VersionedValueFactory valueFactory = new VersionedValue.VersionedValueFactory(partitioner);
ArrayList<Token> endpointTokens = new ArrayList<>();
ArrayList<Token> keyTokens = new ArrayList<>();
- List<InetAddress> hosts = new ArrayList<>();
+ List<InetAddressAndPort> hosts = new ArrayList<>();
List<UUID> hostIds = new ArrayList<>();
for(int i=0; i < RING_SIZE/2; i++)
@@ -221,7 +221,7 @@ public class MoveTest
VersionedValue.VersionedValueFactory valueFactory = new VersionedValue.VersionedValueFactory(partitioner);
ArrayList<Token> endpointTokens = new ArrayList<>();
ArrayList<Token> keyTokens = new ArrayList<>();
- List<InetAddress> hosts = new ArrayList<>();
+ List<InetAddressAndPort> hosts = new ArrayList<>();
List<UUID> hostIds = new ArrayList<>();
for(int i=0; i < RING_SIZE/2; i++)
@@ -312,7 +312,7 @@ public class MoveTest
VersionedValue.VersionedValueFactory valueFactory = new VersionedValue.VersionedValueFactory(partitioner);
ArrayList<Token> endpointTokens = new ArrayList<>();
ArrayList<Token> keyTokens = new ArrayList<>();
- List<InetAddress> hosts = new ArrayList<>();
+ List<InetAddressAndPort> hosts = new ArrayList<>();
List<UUID> hostIds = new ArrayList<>();
Util.createInitialRing(ss, partitioner, endpointTokens, keyTokens, hosts, hostIds, RING_SIZE);
@@ -376,7 +376,7 @@ public class MoveTest
VersionedValue.VersionedValueFactory valueFactory = new VersionedValue.VersionedValueFactory(partitioner);
ArrayList<Token> endpointTokens = new ArrayList<>();
ArrayList<Token> keyTokens = new ArrayList<>();
- List<InetAddress> hosts = new ArrayList<>();
+ List<InetAddressAndPort> hosts = new ArrayList<>();
List<UUID> hostIds = new ArrayList<>();
Util.createInitialRing(ss, partitioner, endpointTokens, keyTokens, hosts, hostIds, RING_SIZE);
@@ -478,38 +478,38 @@ public class MoveTest
finishMove(hosts.get(MOVING_NODE), 30, tmd);
}
- private void moveHost(InetAddress host, int token, TokenMetadata tmd, VersionedValue.VersionedValueFactory valueFactory )
+ private void moveHost(InetAddressAndPort host, int token, TokenMetadata tmd, VersionedValue.VersionedValueFactory valueFactory )
{
StorageService.instance.onChange(host, ApplicationState.STATUS, valueFactory.moving(new BigIntegerToken(String.valueOf(token))));
PendingRangeCalculatorService.instance.blockUntilFinished();
assertTrue(tmd.isMoving(host));
}
- private void finishMove(InetAddress host, int token, TokenMetadata tmd)
+ private void finishMove(InetAddressAndPort host, int token, TokenMetadata tmd)
{
tmd.removeFromMoving(host);
assertTrue(!tmd.isMoving(host));
tmd.updateNormalToken(new BigIntegerToken(String.valueOf(token)), host);
}
- private Map.Entry<Range<Token>, Collection<InetAddress>> generatePendingMapEntry(int start, int end, String... endpoints) throws UnknownHostException
+ private Map.Entry<Range<Token>, Collection<InetAddressAndPort>> generatePendingMapEntry(int start, int end, String... endpoints) throws UnknownHostException
{
- Map<Range<Token>, Collection<InetAddress>> pendingRanges = new HashMap<>();
+ Map<Range<Token>, Collection<InetAddressAndPort>> pendingRanges = new HashMap<>();
pendingRanges.put(generateRange(start, end), makeAddrs(endpoints));
return pendingRanges.entrySet().iterator().next();
}
- private Map<Range<Token>, Collection<InetAddress>> generatePendingRanges(Map.Entry<Range<Token>, Collection<InetAddress>>... entries)
+ private Map<Range<Token>, Collection<InetAddressAndPort>> generatePendingRanges(Map.Entry<Range<Token>, Collection<InetAddressAndPort>>... entries)
{
- Map<Range<Token>, Collection<InetAddress>> pendingRanges = new HashMap<>();
- for(Map.Entry<Range<Token>, Collection<InetAddress>> entry : entries)
+ Map<Range<Token>, Collection<InetAddressAndPort>> pendingRanges = new HashMap<>();
+ for(Map.Entry<Range<Token>, Collection<InetAddressAndPort>> entry : entries)
{
pendingRanges.put(entry.getKey(), entry.getValue());
}
return pendingRanges;
}
- private void assertPendingRanges(TokenMetadata tmd, Map<Range<Token>, Collection<InetAddress>> pendingRanges, String keyspaceName) throws ConfigurationException
+ private void assertPendingRanges(TokenMetadata tmd, Map<Range<Token>, Collection<InetAddressAndPort>> pendingRanges, String keyspaceName) throws ConfigurationException
{
boolean keyspaceFound = false;
for (String nonSystemKeyspaceName : Schema.instance.getNonLocalStrategyKeyspaces())
@@ -523,13 +523,13 @@ public class MoveTest
assert keyspaceFound;
}
- private void assertMaps(Map<Range<Token>, Collection<InetAddress>> expected, PendingRangeMaps actual)
+ private void assertMaps(Map<Range<Token>, Collection<InetAddressAndPort>> expected, PendingRangeMaps actual)
{
int sizeOfActual = 0;
- Iterator<Map.Entry<Range<Token>, List<InetAddress>>> iterator = actual.iterator();
+ Iterator<Map.Entry<Range<Token>, List<InetAddressAndPort>>> iterator = actual.iterator();
while(iterator.hasNext())
{
- Map.Entry<Range<Token>, List<InetAddress>> actualEntry = iterator.next();
+ Map.Entry<Range<Token>, List<InetAddressAndPort>> actualEntry = iterator.next();
assertNotNull(expected.get(actualEntry.getKey()));
assertEquals(new HashSet<>(expected.get(actualEntry.getKey())), new HashSet<>(actualEntry.getValue()));
sizeOfActual++;
@@ -554,15 +554,15 @@ public class MoveTest
ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>();
- List<InetAddress> hosts = new ArrayList<InetAddress>();
+ List<InetAddressAndPort> hosts = new ArrayList<>();
List<UUID> hostIds = new ArrayList<UUID>();
Util.createInitialRing(ss, partitioner, endpointTokens, keyTokens, hosts, hostIds, RING_SIZE);
- Map<Token, List<InetAddress>> expectedEndpoints = new HashMap<Token, List<InetAddress>>();
+ Map<Token, List<InetAddressAndPort>> expectedEndpoints = new HashMap<>();
for (Token token : keyTokens)
{
- List<InetAddress> endpoints = new ArrayList<InetAddress>();
+ List<InetAddressAndPort> endpoints = new ArrayList<>();
Iterator<Token> tokenIter = TokenMetadata.ringIterator(tmd.sortedTokens(), token, false);
while (tokenIter.hasNext())
{
@@ -591,8 +591,8 @@ public class MoveTest
{
int replicationFactor = strategy.getReplicationFactor();
- HashSet<InetAddress> actual = new HashSet<InetAddress>(tmd.getWriteEndpoints(token, keyspaceName, strategy.calculateNaturalEndpoints(token, tmd.cloneOnlyTokenMap())));
- HashSet<InetAddress> expected = new HashSet<InetAddress>();
+ HashSet<InetAddressAndPort> actual = new HashSet<>(tmd.getWriteEndpoints(token, keyspaceName, strategy.calculateNaturalEndpoints(token, tmd.cloneOnlyTokenMap())));
+ HashSet<InetAddressAndPort> expected = new HashSet<>();
for (int i = 0; i < replicationFactor; i++)
{
@@ -628,7 +628,7 @@ public class MoveTest
ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>();
- List<InetAddress> hosts = new ArrayList<InetAddress>();
+ List<InetAddressAndPort> hosts = new ArrayList<>();
List<UUID> hostIds = new ArrayList<UUID>();
// create a ring or 10 nodes
@@ -648,13 +648,13 @@ public class MoveTest
newTokens.put(movingIndex, newToken);
}
- Collection<InetAddress> endpoints;
+ Collection<InetAddressAndPort> endpoints;
tmd = tmd.cloneAfterAllSettled();
ss.setTokenMetadataUnsafe(tmd);
// boot two new nodes with keyTokens.get(5) and keyTokens.get(7)
- InetAddress boot1 = InetAddress.getByName("127.0.1.1");
+ InetAddressAndPort boot1 = InetAddressAndPort.getByName("127.0.1.1");
Gossiper.instance.initializeNodeUnsafe(boot1, UUID.randomUUID(), 1);
Gossiper.instance.injectApplicationState(boot1, ApplicationState.TOKENS, valueFactory.tokens(Collections.singleton(keyTokens.get(5))));
ss.onChange(boot1,
@@ -662,7 +662,7 @@ public class MoveTest
valueFactory.bootstrapping(Collections.<Token>singleton(keyTokens.get(5))));
PendingRangeCalculatorService.instance.blockUntilFinished();
- InetAddress boot2 = InetAddress.getByName("127.0.1.2");
+ InetAddressAndPort boot2 = InetAddressAndPort.getByName("127.0.1.2");
Gossiper.instance.initializeNodeUnsafe(boot2, UUID.randomUUID(), 1);
Gossiper.instance.injectApplicationState(boot2, ApplicationState.TOKENS, valueFactory.tokens(Collections.singleton(keyTokens.get(7))));
ss.onChange(boot2,
@@ -693,35 +693,35 @@ public class MoveTest
* }
*/
- Multimap<InetAddress, Range<Token>> keyspace1ranges = keyspaceStrategyMap.get(Simple_RF1_KeyspaceName).getAddressRanges();
- Collection<Range<Token>> ranges1 = keyspace1ranges.get(InetAddress.getByName("127.0.0.1"));
+ Multimap<InetAddressAndPort, Range<Token>> keyspace1ranges = keyspaceStrategyMap.get(Simple_RF1_KeyspaceName).getAddressRanges();
+ Collection<Range<Token>> ranges1 = keyspace1ranges.get(InetAddressAndPort.getByName("127.0.0.1"));
assertEquals(1, collectionSize(ranges1));
assertEquals(generateRange(97, 0), ranges1.iterator().next());
- Collection<Range<Token>> ranges2 = keyspace1ranges.get(InetAddress.getByName("127.0.0.2"));
+ Collection<Range<Token>> ranges2 = keyspace1ranges.get(InetAddressAndPort.getByName("127.0.0.2"));
assertEquals(1, collectionSize(ranges2));
assertEquals(generateRange(0, 10), ranges2.iterator().next());
- Collection<Range<Token>> ranges3 = keyspace1ranges.get(InetAddress.getByName("127.0.0.3"));
+ Collection<Range<Token>> ranges3 = keyspace1ranges.get(InetAddressAndPort.getByName("127.0.0.3"));
assertEquals(1, collectionSize(ranges3));
assertEquals(generateRange(10, 20), ranges3.iterator().next());
- Collection<Range<Token>> ranges4 = keyspace1ranges.get(InetAddress.getByName("127.0.0.4"));
+ Collection<Range<Token>> ranges4 = keyspace1ranges.get(InetAddressAndPort.getByName("127.0.0.4"));
assertEquals(1, collectionSize(ranges4));
assertEquals(generateRange(20, 30), ranges4.iterator().next());
- Collection<Range<Token>> ranges5 = keyspace1ranges.get(InetAddress.getByName("127.0.0.5"));
+ Collection<Range<Token>> ranges5 = keyspace1ranges.get(InetAddressAndPort.getByName("127.0.0.5"));
assertEquals(1, collectionSize(ranges5));
assertEquals(generateRange(30, 40), ranges5.iterator().next());
- Collection<Range<Token>> ranges6 = keyspace1ranges.get(InetAddress.getByName("127.0.0.6"));
+ Collection<Range<Token>> ranges6 = keyspace1ranges.get(InetAddressAndPort.getByName("127.0.0.6"));
assertEquals(1, collectionSize(ranges6));
assertEquals(generateRange(40, 50), ranges6.iterator().next());
- Collection<Range<Token>> ranges7 = keyspace1ranges.get(InetAddress.getByName("127.0.0.7"));
+ Collection<Range<Token>> ranges7 = keyspace1ranges.get(InetAddressAndPort.getByName("127.0.0.7"));
assertEquals(1, collectionSize(ranges7));
assertEquals(generateRange(50, 67), ranges7.iterator().next());
- Collection<Range<Token>> ranges8 = keyspace1ranges.get(InetAddress.getByName("127.0.0.8"));
+ Collection<Range<Token>> ranges8 = keyspace1ranges.get(InetAddressAndPort.getByName("127.0.0.8"));
assertEquals(1, collectionSize(ranges8));
assertEquals(generateRange(67, 70), ranges8.iterator().next());
- Collection<Range<Token>> ranges9 = keyspace1ranges.get(InetAddress.getByName("127.0.0.9"));
+ Collection<Range<Token>> ranges9 = keyspace1ranges.get(InetAddressAndPort.getByName("127.0.0.9"));
assertEquals(1, collectionSize(ranges9));
assertEquals(generateRange(70, 87), ranges9.iterator().next());
- Collection<Range<Token>> ranges10 = keyspace1ranges.get(InetAddress.getByName("127.0.0.10"));
+ Collection<Range<Token>> ranges10 = keyspace1ranges.get(InetAddressAndPort.getByName("127.0.0.10"));
assertEquals(1, collectionSize(ranges10));
assertEquals(generateRange(87, 97), ranges10.iterator().next());
@@ -742,35 +742,35 @@ public class MoveTest
* }
*/
- Multimap<InetAddress, Range<Token>> keyspace3ranges = keyspaceStrategyMap.get(KEYSPACE3).getAddressRanges();
- ranges1 = keyspace3ranges.get(InetAddress.getByName("127.0.0.1"));
+ Multimap<InetAddressAndPort, Range<Token>> keyspace3ranges = keyspaceStrategyMap.get(KEYSPACE3).getAddressRanges();
+ ranges1 = keyspace3ranges.get(InetAddressAndPort.getByName("127.0.0.1"));
assertEquals(collectionSize(ranges1), 5);
assertTrue(ranges1.equals(generateRanges(97, 0, 70, 87, 50, 67, 87, 97, 67, 70)));
- ranges2 = keyspace3ranges.get(InetAddress.getByName("127.0.0.2"));
+ ranges2 = keyspace3ranges.get(InetAddressAndPort.getByName("127.0.0.2"));
assertEquals(collectionSize(ranges2), 5);
assertTrue(ranges2.equals(generateRanges(97, 0, 70, 87, 87, 97, 0, 10, 67, 70)));
- ranges3 = keyspace3ranges.get(InetAddress.getByName("127.0.0.3"));
+ ranges3 = keyspace3ranges.get(InetAddressAndPort.getByName("127.0.0.3"));
assertEquals(collectionSize(ranges3), 5);
assertTrue(ranges3.equals(generateRanges(97, 0, 70, 87, 87, 97, 0, 10, 10, 20)));
- ranges4 = keyspace3ranges.get(InetAddress.getByName("127.0.0.4"));
+ ranges4 = keyspace3ranges.get(InetAddressAndPort.getByName("127.0.0.4"));
assertEquals(collectionSize(ranges4), 5);
assertTrue(ranges4.equals(generateRanges(97, 0, 20, 30, 87, 97, 0, 10, 10, 20)));
- ranges5 = keyspace3ranges.get(InetAddress.getByName("127.0.0.5"));
+ ranges5 = keyspace3ranges.get(InetAddressAndPort.getByName("127.0.0.5"));
assertEquals(collectionSize(ranges5), 5);
assertTrue(ranges5.equals(generateRanges(97, 0, 30, 40, 20, 30, 0, 10, 10, 20)));
- ranges6 = keyspace3ranges.get(InetAddress.getByName("127.0.0.6"));
+ ranges6 = keyspace3ranges.get(InetAddressAndPort.getByName("127.0.0.6"));
assertEquals(collectionSize(ranges6), 5);
assertTrue(ranges6.equals(generateRanges(40, 50, 30, 40, 20, 30, 0, 10, 10, 20)));
- ranges7 = keyspace3ranges.get(InetAddress.getByName("127.0.0.7"));
+ ranges7 = keyspace3ranges.get(InetAddressAndPort.getByName("127.0.0.7"));
assertEquals(collectionSize(ranges7), 5);
assertTrue(ranges7.equals(generateRanges(40, 50, 30, 40, 50, 67, 20, 30, 10, 20)));
- ranges8 = keyspace3ranges.get(InetAddress.getByName("127.0.0.8"));
+ ranges8 = keyspace3ranges.get(InetAddressAndPort.getByName("127.0.0.8"));
assertEquals(collectionSize(ranges8), 5);
assertTrue(ranges8.equals(generateRanges(40, 50, 30, 40, 50, 67, 20, 30, 67, 70)));
- ranges9 = keyspace3ranges.get(InetAddress.getByName("127.0.0.9"));
+ ranges9 = keyspace3ranges.get(InetAddressAndPort.getByName("127.0.0.9"));
assertEquals(collectionSize(ranges9), 5);
assertTrue(ranges9.equals(generateRanges(40, 50, 70, 87, 30, 40, 50, 67, 67, 70)));
- ranges10 = keyspace3ranges.get(InetAddress.getByName("127.0.0.10"));
+ ranges10 = keyspace3ranges.get(InetAddressAndPort.getByName("127.0.0.10"));
assertEquals(collectionSize(ranges10), 5);
assertTrue(ranges10.equals(generateRanges(40, 50, 70, 87, 50, 67, 87, 97, 67, 70)));
@@ -790,41 +790,41 @@ public class MoveTest
* /127.0.0.10=[(70,87], (87,97], (67,70]]
* }
*/
- Multimap<InetAddress, Range<Token>> keyspace4ranges = keyspaceStrategyMap.get(Simple_RF3_KeyspaceName).getAddressRanges();
- ranges1 = keyspace4ranges.get(InetAddress.getByName("127.0.0.1"));
+ Multimap<InetAddressAndPort, Range<Token>> keyspace4ranges = keyspaceStrategyMap.get(Simple_RF3_KeyspaceName).getAddressRanges();
+ ranges1 = keyspace4ranges.get(InetAddressAndPort.getByName("127.0.0.1"));
assertEquals(collectionSize(ranges1), 3);
assertTrue(ranges1.equals(generateRanges(97, 0, 70, 87, 87, 97)));
- ranges2 = keyspace4ranges.get(InetAddress.getByName("127.0.0.2"));
+ ranges2 = keyspace4ranges.get(InetAddressAndPort.getByName("127.0.0.2"));
assertEquals(collectionSize(ranges2), 3);
assertTrue(ranges2.equals(generateRanges(97, 0, 87, 97, 0, 10)));
- ranges3 = keyspace4ranges.get(InetAddress.getByName("127.0.0.3"));
+ ranges3 = keyspace4ranges.get(InetAddressAndPort.getByName("127.0.0.3"));
assertEquals(collectionSize(ranges3), 3);
assertTrue(ranges3.equals(generateRanges(97, 0, 0, 10, 10, 20)));
- ranges4 = keyspace4ranges.get(InetAddress.getByName("127.0.0.4"));
+ ranges4 = keyspace4ranges.get(InetAddressAndPort.getByName("127.0.0.4"));
assertEquals(collectionSize(ranges4), 3);
assertTrue(ranges4.equals(generateRanges(20, 30, 0, 10, 10, 20)));
- ranges5 = keyspace4ranges.get(InetAddress.getByName("127.0.0.5"));
+ ranges5 = keyspace4ranges.get(InetAddressAndPort.getByName("127.0.0.5"));
assertEquals(collectionSize(ranges5), 3);
assertTrue(ranges5.equals(generateRanges(30, 40, 20, 30, 10, 20)));
- ranges6 = keyspace4ranges.get(InetAddress.getByName("127.0.0.6"));
+ ranges6 = keyspace4ranges.get(InetAddressAndPort.getByName("127.0.0.6"));
assertEquals(collectionSize(ranges6), 3);
assertTrue(ranges6.equals(generateRanges(40, 50, 30, 40, 20, 30)));
- ranges7 = keyspace4ranges.get(InetAddress.getByName("127.0.0.7"));
+ ranges7 = keyspace4ranges.get(InetAddressAndPort.getByName("127.0.0.7"));
assertEquals(collectionSize(ranges7), 3);
assertTrue(ranges7.equals(generateRanges(40, 50, 30, 40, 50, 67)));
- ranges8 = keyspace4ranges.get(InetAddress.getByName("127.0.0.8"));
+ ranges8 = keyspace4ranges.get(InetAddressAndPort.getByName("127.0.0.8"));
assertEquals(collectionSize(ranges8), 3);
assertTrue(ranges8.equals(generateRanges(40, 50, 50, 67, 67, 70)));
- ranges9 = keyspace4ranges.get(InetAddress.getByName("127.0.0.9"));
+ ranges9 = keyspace4ranges.get(InetAddressAndPort.getByName("127.0.0.9"));
assertEquals(collectionSize(ranges9), 3);
assertTrue(ranges9.equals(generateRanges(70, 87, 50, 67, 67, 70)));
- ranges10 = keyspace4ranges.get(InetAddress.getByName("127.0.0.10"));
+ ranges10 = keyspace4ranges.get(InetAddressAndPort.getByName("127.0.0.10"));
assertEquals(collectionSize(ranges10), 3);
assertTrue(ranges10.equals(generateRanges(70, 87, 87, 97, 67, 70)));
// pre-calculate the results.
- Map<String, Multimap<Token, InetAddress>> expectedEndpoints = new HashMap<String, Multimap<Token, InetAddress>>();
- expectedEndpoints.put(Simple_RF1_KeyspaceName, HashMultimap.<Token, InetAddress>create());
+ Map<String, Multimap<Token, InetAddressAndPort>> expectedEndpoints = new HashMap<>();
+ expectedEndpoints.put(Simple_RF1_KeyspaceName, HashMultimap.<Token, InetAddressAndPort>create());
expectedEndpoints.get(Simple_RF1_KeyspaceName).putAll(new BigIntegerToken("5"), makeAddrs("127.0.0.2"));
expectedEndpoints.get(Simple_RF1_KeyspaceName).putAll(new BigIntegerToken("15"), makeAddrs("127.0.0.3"));
expectedEndpoints.get(Simple_RF1_KeyspaceName).putAll(new BigIntegerToken("25"), makeAddrs("127.0.0.4"));
@@ -835,7 +835,7 @@ public class MoveTest
expectedEndpoints.get(Simple_RF1_KeyspaceName).putAll(new BigIntegerToken("75"), makeAddrs("127.0.0.9", "127.0.1.2"));
expectedEndpoints.get(Simple_RF1_KeyspaceName).putAll(new BigIntegerToken("85"), makeAddrs("127.0.0.9"));
expectedEndpoints.get(Simple_RF1_KeyspaceName).putAll(new BigIntegerToken("95"), makeAddrs("127.0.0.10"));
- expectedEndpoints.put(KEYSPACE2, HashMultimap.<Token, InetAddress>create());
+ expectedEndpoints.put(KEYSPACE2, HashMultimap.<Token, InetAddressAndPort>create());
expectedEndpoints.get(KEYSPACE2).putAll(new BigIntegerToken("5"), makeAddrs("127.0.0.2"));
expectedEndpoints.get(KEYSPACE2).putAll(new BigIntegerToken("15"), makeAddrs("127.0.0.3"));
expectedEndpoints.get(KEYSPACE2).putAll(new BigIntegerToken("25"), makeAddrs("127.0.0.4"));
@@ -846,7 +846,7 @@ public class MoveTest
expectedEndpoints.get(KEYSPACE2).putAll(new BigIntegerToken("75"), makeAddrs("127.0.0.9", "127.0.1.2"));
expectedEndpoints.get(KEYSPACE2).putAll(new BigIntegerToken("85"), makeAddrs("127.0.0.9"));
expectedEndpoints.get(KEYSPACE2).putAll(new BigIntegerToken("95"), makeAddrs("127.0.0.10"));
- expectedEndpoints.put(KEYSPACE3, HashMultimap.<Token, InetAddress>create());
+ expectedEndpoints.put(KEYSPACE3, HashMultimap.<Token, InetAddressAndPort>create());
expectedEndpoints.get(KEYSPACE3).putAll(new BigIntegerToken("5"), makeAddrs("127.0.0.2", "127.0.0.3", "127.0.0.4", "127.0.0.5", "127.0.0.6"));
expectedEndpoints.get(KEYSPACE3).putAll(new BigIntegerToken("15"), makeAddrs("127.0.0.3", "127.0.0.4", "127.0.0.5", "127.0.0.6", "127.0.0.7", "127.0.1.1"));
expectedEndpoints.get(KEYSPACE3).putAll(new BigIntegerToken("25"), makeAddrs("127.0.0.4", "127.0.0.5", "127.0.0.6", "127.0.0.7", "127.0.0.8", "127.0.1.1"));
@@ -857,7 +857,7 @@ public class MoveTest
expectedEndpoints.get(KEYSPACE3).putAll(new BigIntegerToken("75"), makeAddrs("127.0.0.9", "127.0.0.10", "127.0.0.1", "127.0.0.2", "127.0.0.3", "127.0.1.2"));
expectedEndpoints.get(KEYSPACE3).putAll(new BigIntegerToken("85"), makeAddrs("127.0.0.9", "127.0.0.10", "127.0.0.1", "127.0.0.2", "127.0.0.3"));
expectedEndpoints.get(KEYSPACE3).putAll(new BigIntegerToken("95"), makeAddrs("127.0.0.10", "127.0.0.1", "127.0.0.2", "127.0.0.3", "127.0.0.4"));
- expectedEndpoints.put(Simple_RF3_KeyspaceName, HashMultimap.<Token, InetAddress>create());
+ expectedEndpoints.put(Simple_RF3_KeyspaceName, HashMultimap.<Token, InetAddressAndPort>create());
expectedEndpoints.get(Simple_RF3_KeyspaceName).putAll(new BigIntegerToken("5"), makeAddrs("127.0.0.2", "127.0.0.3", "127.0.0.4"));
expectedEndpoints.get(Simple_RF3_KeyspaceName).putAll(new BigIntegerToken("15"), makeAddrs("127.0.0.3", "127.0.0.4", "127.0.0.5"));
expectedEndpoints.get(Simple_RF3_KeyspaceName).putAll(new BigIntegerToken("25"), makeAddrs("127.0.0.4", "127.0.0.5", "127.0.0.6"));
@@ -970,7 +970,7 @@ public class MoveTest
ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>();
- List<InetAddress> hosts = new ArrayList<InetAddress>();
+ List<InetAddressAndPort> hosts = new ArrayList<>();
List<UUID> hostIds = new ArrayList<UUID>();
// create a ring or 6 nodes
@@ -1001,11 +1001,11 @@ public class MoveTest
assertEquals(newToken, tmd.getToken(hosts.get(2)));
}
- private static Collection<InetAddress> makeAddrs(String... hosts) throws UnknownHostException
+ private static Collection<InetAddressAndPort> makeAddrs(String... hosts) throws UnknownHostException
{
- ArrayList<InetAddress> addrs = new ArrayList<InetAddress>(hosts.length);
+ ArrayList<InetAddressAndPort> addrs = new ArrayList<>(hosts.length);
for (String host : hosts)
- addrs.add(InetAddress.getByName(host));
+ addrs.add(InetAddressAndPort.getByName(host));
return addrs;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/test/unit/org/apache/cassandra/service/ProtocolBetaVersionTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/service/ProtocolBetaVersionTest.java b/test/unit/org/apache/cassandra/service/ProtocolBetaVersionTest.java
index 4ade4ad..a7551f4 100644
--- a/test/unit/org/apache/cassandra/service/ProtocolBetaVersionTest.java
+++ b/test/unit/org/apache/cassandra/service/ProtocolBetaVersionTest.java
@@ -110,7 +110,7 @@ public class ProtocolBetaVersionTest extends CQLTester
}
catch (Exception e)
{
- assertEquals("Beta version of server used (5/v5-beta), but USE_BETA flag is not set",
+ assertEquals("Beta version of server used (6/v6-beta), but USE_BETA flag is not set",
e.getMessage());
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/test/unit/org/apache/cassandra/service/ReadExecutorTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/service/ReadExecutorTest.java b/test/unit/org/apache/cassandra/service/ReadExecutorTest.java
index 7630cc6..f21e241 100644
--- a/test/unit/org/apache/cassandra/service/ReadExecutorTest.java
+++ b/test/unit/org/apache/cassandra/service/ReadExecutorTest.java
@@ -18,7 +18,6 @@
package org.apache.cassandra.service;
-import java.net.InetAddress;
import java.util.List;
import java.util.concurrent.TimeUnit;
@@ -36,6 +35,7 @@ import org.apache.cassandra.db.SinglePartitionReadCommand;
import org.apache.cassandra.exceptions.ReadFailureException;
import org.apache.cassandra.exceptions.ReadTimeoutException;
import org.apache.cassandra.exceptions.RequestFailureReason;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.MessageOut;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.KeyspaceParams;
@@ -47,7 +47,7 @@ public class ReadExecutorTest
{
static Keyspace ks;
static ColumnFamilyStore cfs;
- static List<InetAddress> targets;
+ static List<InetAddressAndPort> targets;
@BeforeClass
public static void setUpClass() throws Throwable
@@ -56,7 +56,7 @@ public class ReadExecutorTest
SchemaLoader.createKeyspace("Foo", KeyspaceParams.simple(3), SchemaLoader.standardCFMD("Foo", "Bar"));
ks = Keyspace.open("Foo");
cfs = ks.getColumnFamilyStore("Bar");
- targets = ImmutableList.of(InetAddress.getByName("127.0.0.255"), InetAddress.getByName("127.0.0.254"), InetAddress.getByName("127.0.0.253"));
+ targets = ImmutableList.of(InetAddressAndPort.getByName("127.0.0.255"), InetAddressAndPort.getByName("127.0.0.254"), InetAddressAndPort.getByName("127.0.0.253"));
cfs.sampleLatencyNanos = 0;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/test/unit/org/apache/cassandra/service/RemoveTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/service/RemoveTest.java b/test/unit/org/apache/cassandra/service/RemoveTest.java
index afc5b25..6714a83 100644
--- a/test/unit/org/apache/cassandra/service/RemoveTest.java
+++ b/test/unit/org/apache/cassandra/service/RemoveTest.java
@@ -20,7 +20,6 @@
package org.apache.cassandra.service;
import java.io.IOException;
-import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -40,6 +39,7 @@ import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.gms.ApplicationState;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.gms.VersionedValue.VersionedValueFactory;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.net.MessageOut;
import org.apache.cassandra.net.MessagingService;
@@ -61,9 +61,9 @@ public class RemoveTest
static IPartitioner oldPartitioner;
ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>();
- List<InetAddress> hosts = new ArrayList<InetAddress>();
+ List<InetAddressAndPort> hosts = new ArrayList<>();
List<UUID> hostIds = new ArrayList<UUID>();
- InetAddress removalhost;
+ InetAddressAndPort removalhost;
UUID removalId;
@BeforeClass
@@ -120,7 +120,7 @@ public class RemoveTest
VersionedValueFactory valueFactory = new VersionedValueFactory(DatabaseDescriptor.getPartitioner());
Collection<Token> tokens = Collections.singleton(DatabaseDescriptor.getPartitioner().getRandomToken());
- InetAddress joininghost = hosts.get(4);
+ InetAddressAndPort joininghost = hosts.get(4);
UUID joiningId = hostIds.get(4);
hosts.remove(joininghost);
@@ -159,10 +159,10 @@ public class RemoveTest
assertTrue(tmd.isLeaving(removalhost));
assertEquals(1, tmd.getSizeOfLeavingEndpoints());
- for (InetAddress host : hosts)
+ for (InetAddressAndPort host : hosts)
{
- MessageOut msg = new MessageOut(host, MessagingService.Verb.REPLICATION_FINISHED, null, null, Collections.<String, byte[]>emptyMap());
- MessagingService.instance().sendRR(msg, FBUtilities.getBroadcastAddress());
+ MessageOut msg = new MessageOut(host, MessagingService.Verb.REPLICATION_FINISHED, null, null, Collections.<Object>emptyList());
+ MessagingService.instance().sendRR(msg, FBUtilities.getBroadcastAddressAndPort());
}
remover.join();
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/test/unit/org/apache/cassandra/service/SerializationsTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/service/SerializationsTest.java b/test/unit/org/apache/cassandra/service/SerializationsTest.java
index d943bb7..c29e7a8 100644
--- a/test/unit/org/apache/cassandra/service/SerializationsTest.java
+++ b/test/unit/org/apache/cassandra/service/SerializationsTest.java
@@ -19,7 +19,6 @@
package org.apache.cassandra.service;
import java.io.IOException;
-import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -41,6 +40,7 @@ import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.util.DataInputPlus.DataInputStreamPlus;
import org.apache.cassandra.io.util.DataOutputStreamPlus;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.MessageIn;
import org.apache.cassandra.repair.NodePair;
import org.apache.cassandra.repair.RepairJobDesc;
@@ -61,6 +61,8 @@ public class SerializationsTest extends AbstractSerializationsTester
private static Range<Token> FULL_RANGE;
private static RepairJobDesc DESC;
+ private static final int PORT = 7010;
+
@BeforeClass
public static void defineSchema() throws Exception
{
@@ -123,7 +125,7 @@ public class SerializationsTest extends AbstractSerializationsTester
// empty validation
mt.addMerkleTree((int) Math.pow(2, 15), FULL_RANGE);
- Validator v0 = new Validator(DESC, FBUtilities.getBroadcastAddress(), -1, PreviewKind.NONE);
+ Validator v0 = new Validator(DESC, FBUtilities.getBroadcastAddressAndPort(), -1, PreviewKind.NONE);
ValidationComplete c0 = new ValidationComplete(DESC, mt);
// validation with a tree
@@ -131,7 +133,7 @@ public class SerializationsTest extends AbstractSerializationsTester
mt.addMerkleTree(Integer.MAX_VALUE, FULL_RANGE);
for (int i = 0; i < 10; i++)
mt.split(p.getRandomToken());
- Validator v1 = new Validator(DESC, FBUtilities.getBroadcastAddress(), -1, PreviewKind.NONE);
+ Validator v1 = new Validator(DESC, FBUtilities.getBroadcastAddressAndPort(), -1, PreviewKind.NONE);
ValidationComplete c1 = new ValidationComplete(DESC, mt);
// validation failed
@@ -180,9 +182,9 @@ public class SerializationsTest extends AbstractSerializationsTester
private void testSyncRequestWrite() throws IOException
{
- InetAddress local = InetAddress.getByAddress(new byte[]{127, 0, 0, 1});
- InetAddress src = InetAddress.getByAddress(new byte[]{127, 0, 0, 2});
- InetAddress dest = InetAddress.getByAddress(new byte[]{127, 0, 0, 3});
+ InetAddressAndPort local = InetAddressAndPort.getByNameOverrideDefaults("127.0.0.1", PORT);
+ InetAddressAndPort src = InetAddressAndPort.getByNameOverrideDefaults("127.0.0.2", PORT);
+ InetAddressAndPort dest = InetAddressAndPort.getByNameOverrideDefaults("127.0.0.3", PORT);
SyncRequest message = new SyncRequest(DESC, local, src, dest, Collections.singleton(FULL_RANGE), PreviewKind.NONE);
testRepairMessageWrite("service.SyncRequest.bin", message);
@@ -194,9 +196,9 @@ public class SerializationsTest extends AbstractSerializationsTester
if (EXECUTE_WRITES)
testSyncRequestWrite();
- InetAddress local = InetAddress.getByAddress(new byte[]{127, 0, 0, 1});
- InetAddress src = InetAddress.getByAddress(new byte[]{127, 0, 0, 2});
- InetAddress dest = InetAddress.getByAddress(new byte[]{127, 0, 0, 3});
+ InetAddressAndPort local = InetAddressAndPort.getByNameOverrideDefaults("127.0.0.1", PORT);
+ InetAddressAndPort src = InetAddressAndPort.getByNameOverrideDefaults("127.0.0.2", PORT);
+ InetAddressAndPort dest = InetAddressAndPort.getByNameOverrideDefaults("127.0.0.3", PORT);
try (DataInputStreamPlus in = getInput("service.SyncRequest.bin"))
{
@@ -214,8 +216,8 @@ public class SerializationsTest extends AbstractSerializationsTester
private void testSyncCompleteWrite() throws IOException
{
- InetAddress src = InetAddress.getByAddress(new byte[]{127, 0, 0, 2});
- InetAddress dest = InetAddress.getByAddress(new byte[]{127, 0, 0, 3});
+ InetAddressAndPort src = InetAddressAndPort.getByNameOverrideDefaults("127.0.0.2", PORT);
+ InetAddressAndPort dest = InetAddressAndPort.getByNameOverrideDefaults("127.0.0.3", PORT);
// sync success
List<SessionSummary> summaries = new ArrayList<>();
summaries.add(new SessionSummary(src, dest,
@@ -235,8 +237,8 @@ public class SerializationsTest extends AbstractSerializationsTester
if (EXECUTE_WRITES)
testSyncCompleteWrite();
- InetAddress src = InetAddress.getByAddress(new byte[]{127, 0, 0, 2});
- InetAddress dest = InetAddress.getByAddress(new byte[]{127, 0, 0, 3});
+ InetAddressAndPort src = InetAddressAndPort.getByNameOverrideDefaults("127.0.0.2", PORT);
+ InetAddressAndPort dest = InetAddressAndPort.getByNameOverrideDefaults("127.0.0.3", PORT);
NodePair nodes = new NodePair(src, dest);
try (DataInputStreamPlus in = getInput("service.SyncComplete.bin"))
@@ -246,6 +248,8 @@ public class SerializationsTest extends AbstractSerializationsTester
assert message.messageType == RepairMessage.Type.SYNC_COMPLETE;
assert DESC.equals(message.desc);
+ System.out.println(nodes);
+ System.out.println(((SyncComplete) message).nodes);
assert nodes.equals(((SyncComplete) message).nodes);
assert ((SyncComplete) message).success;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/test/unit/org/apache/cassandra/service/StorageProxyTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/service/StorageProxyTest.java b/test/unit/org/apache/cassandra/service/StorageProxyTest.java
index bdf45fe..590cfeb 100644
--- a/test/unit/org/apache/cassandra/service/StorageProxyTest.java
+++ b/test/unit/org/apache/cassandra/service/StorageProxyTest.java
@@ -18,7 +18,6 @@
*/
package org.apache.cassandra.service;
-import java.net.InetAddress;
import java.util.List;
import org.junit.BeforeClass;
@@ -27,6 +26,7 @@ import org.junit.Test;
import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.*;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.TokenMetadata;
import static org.apache.cassandra.Util.rp;
@@ -81,8 +81,8 @@ public class StorageProxyTest
DatabaseDescriptor.daemonInitialization();
DatabaseDescriptor.getHintsDirectory().mkdir();
TokenMetadata tmd = StorageService.instance.getTokenMetadata();
- tmd.updateNormalToken(token("1"), InetAddress.getByName("127.0.0.1"));
- tmd.updateNormalToken(token("6"), InetAddress.getByName("127.0.0.6"));
+ tmd.updateNormalToken(token("1"), InetAddressAndPort.getByName("127.0.0.1"));
+ tmd.updateNormalToken(token("6"), InetAddressAndPort.getByName("127.0.0.6"));
}
// test getRestrictedRanges for token
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org