diff --git a/src/java/org/apache/cassandra/gms/EndpointState.java b/src/java/org/apache/cassandra/gms/EndpointState.java index 0e6985aba6..931da8d7a9 100644 --- a/src/java/org/apache/cassandra/gms/EndpointState.java +++ b/src/java/org/apache/cassandra/gms/EndpointState.java @@ -18,7 +18,11 @@ package org.apache.cassandra.gms; import java.io.*; +import java.util.Collections; +import java.util.EnumMap; import java.util.Map; +import java.util.Set; +import java.util.concurrent.atomic.AtomicReference; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -27,8 +31,6 @@ import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.io.IVersionedSerializer; import org.apache.cassandra.io.util.DataOutputPlus; -import org.cliffc.high_scale_lib.NonBlockingHashMap; - /** * This abstraction represents both the HeartBeatState and the ApplicationState in an EndpointState * instance. Any state for a given endpoint can be retrieved from this instance. @@ -42,15 +44,21 @@ public class EndpointState public final static IVersionedSerializer serializer = new EndpointStateSerializer(); private volatile HeartBeatState hbState; - final Map applicationState = new NonBlockingHashMap(); + private final AtomicReference> applicationState; /* fields below do not get serialized */ private volatile long updateTimestamp; private volatile boolean isAlive; EndpointState(HeartBeatState initialHbState) + { + this(initialHbState, new EnumMap(ApplicationState.class)); + } + + EndpointState(HeartBeatState initialHbState, Map states) { hbState = initialHbState; + applicationState = new AtomicReference>(new EnumMap<>(states)); updateTimestamp = System.nanoTime(); isAlive = true; } @@ -68,21 +76,37 @@ public class EndpointState public VersionedValue getApplicationState(ApplicationState key) { - return applicationState.get(key); + return applicationState.get().get(key); } - /** - * TODO replace this with operations that don't expose private state - */ - @Deprecated - public Map getApplicationStateMap() + public Set> states() { - return applicationState; + return applicationState.get().entrySet(); } - void addApplicationState(ApplicationState key, VersionedValue value) + public void addApplicationState(ApplicationState key, VersionedValue value) { - applicationState.put(key, value); + addApplicationStates(Collections.singletonMap(key, value)); + } + + public void addApplicationStates(Map values) + { + addApplicationStates(values.entrySet()); + } + + public void addApplicationStates(Set> values) + { + while (true) + { + Map orig = applicationState.get(); + Map copy = new EnumMap<>(orig); + + for (Map.Entry value : values) + copy.put(value.getKey(), value.getValue()); + + if (applicationState.compareAndSet(orig, copy)) + return; + } } /* getters and setters */ @@ -133,7 +157,7 @@ public class EndpointState public String toString() { - return "EndpointState: HeartBeatState = " + hbState + ", AppStateMap = " + applicationState; + return "EndpointState: HeartBeatState = " + hbState + ", AppStateMap = " + applicationState.get(); } } @@ -146,12 +170,12 @@ class EndpointStateSerializer implements IVersionedSerializer HeartBeatState.serializer.serialize(hbState, out, version); /* serialize the map of ApplicationState objects */ - int size = epState.applicationState.size(); - out.writeInt(size); - for (Map.Entry entry : epState.applicationState.entrySet()) + Set> states = epState.states(); + out.writeInt(states.size()); + for (Map.Entry state : states) { - VersionedValue value = entry.getValue(); - out.writeInt(entry.getKey().ordinal()); + VersionedValue value = state.getValue(); + out.writeInt(state.getKey().ordinal()); VersionedValue.serializer.serialize(value, out, version); } } @@ -159,26 +183,28 @@ class EndpointStateSerializer implements IVersionedSerializer public EndpointState deserialize(DataInput in, int version) throws IOException { HeartBeatState hbState = HeartBeatState.serializer.deserialize(in, version); - EndpointState epState = new EndpointState(hbState); int appStateSize = in.readInt(); + Map states = new EnumMap<>(ApplicationState.class); for (int i = 0; i < appStateSize; ++i) { int key = in.readInt(); VersionedValue value = VersionedValue.serializer.deserialize(in, version); - epState.addApplicationState(Gossiper.STATES[key], value); + states.put(Gossiper.STATES[key], value); } - return epState; + + return new EndpointState(hbState, states); } public long serializedSize(EndpointState epState, int version) { long size = HeartBeatState.serializer.serializedSize(epState.getHeartBeatState(), version); - size += TypeSizes.NATIVE.sizeof(epState.applicationState.size()); - for (Map.Entry entry : epState.applicationState.entrySet()) + Set> states = epState.states(); + size += TypeSizes.NATIVE.sizeof(states.size()); + for (Map.Entry state : states) { - VersionedValue value = entry.getValue(); - size += TypeSizes.NATIVE.sizeof(entry.getKey().ordinal()); + VersionedValue value = state.getValue(); + size += TypeSizes.NATIVE.sizeof(state.getKey().ordinal()); size += VersionedValue.serializer.serializedSize(value, version); } return size; diff --git a/src/java/org/apache/cassandra/gms/FailureDetector.java b/src/java/org/apache/cassandra/gms/FailureDetector.java index c563872d19..a0754b1044 100644 --- a/src/java/org/apache/cassandra/gms/FailureDetector.java +++ b/src/java/org/apache/cassandra/gms/FailureDetector.java @@ -192,15 +192,16 @@ public class FailureDetector implements IFailureDetector, FailureDetectorMBean { sb.append(" generation:").append(endpointState.getHeartBeatState().getGeneration()).append("\n"); sb.append(" heartbeat:").append(endpointState.getHeartBeatState().getHeartBeatVersion()).append("\n"); - for (Map.Entry state : endpointState.applicationState.entrySet()) + for (Map.Entry state : endpointState.states()) { if (state.getKey() == ApplicationState.TOKENS) continue; sb.append(" ").append(state.getKey()).append(":").append(state.getValue().version).append(":").append(state.getValue().value).append("\n"); } - if (endpointState.applicationState.containsKey(ApplicationState.TOKENS)) + VersionedValue tokens = endpointState.getApplicationState(ApplicationState.TOKENS); + if (tokens != null) { - sb.append(" TOKENS:").append(endpointState.applicationState.get(ApplicationState.TOKENS).version).append(":\n"); + sb.append(" TOKENS:").append(tokens.version).append(":\n"); } else { diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index f78dc7a232..86fdab23b0 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -224,7 +224,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean return true; try { - if (entry.getValue().getApplicationStateMap().containsKey(ApplicationState.INTERNAL_IP) && seeds.contains(InetAddress.getByName(entry.getValue().getApplicationState(ApplicationState.INTERNAL_IP).value))) + VersionedValue internalIp = entry.getValue().getApplicationState(ApplicationState.INTERNAL_IP); + if (internalIp != null && seeds.contains(InetAddress.getByName(internalIp.value))) return true; } catch (UnknownHostException e) @@ -371,8 +372,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean int getMaxEndpointStateVersion(EndpointState epState) { int maxVersion = epState.getHeartBeatState().getHeartBeatVersion(); - for (VersionedValue value : epState.getApplicationStateMap().values()) - maxVersion = Math.max(maxVersion, value.version); + for (Map.Entry state : epState.states()) + maxVersion = Math.max(maxVersion, state.getValue().version); return maxVersion; } @@ -525,8 +526,10 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean logger.info("Advertising removal for {}", endpoint); epState.updateTimestamp(); // make sure we don't evict it too soon epState.getHeartBeatState().forceNewerGenerationUnsafe(); - epState.addApplicationState(ApplicationState.STATUS, StorageService.instance.valueFactory.removingNonlocal(hostId)); - epState.addApplicationState(ApplicationState.REMOVAL_COORDINATOR, StorageService.instance.valueFactory.removalCoordinator(localHostId)); + Map states = new EnumMap<>(ApplicationState.class); + states.put(ApplicationState.STATUS, StorageService.instance.valueFactory.removingNonlocal(hostId)); + states.put(ApplicationState.REMOVAL_COORDINATOR, StorageService.instance.valueFactory.removalCoordinator(localHostId)); + epState.addApplicationStates(states); endpointStateMap.put(endpoint, epState); } @@ -853,7 +856,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean logger.trace("local heartbeat version {} greater than {} for {}", localHbVersion, version, forEndpoint); } /* Accumulate all application states whose versions are greater than "version" variable */ - for (Entry entry : epState.getApplicationStateMap().entrySet()) + Map states = new EnumMap<>(ApplicationState.class); + for (Entry entry : epState.states()) { VersionedValue value = entry.getValue(); if (value.version > version) @@ -865,9 +869,11 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean final ApplicationState key = entry.getKey(); if (logger.isTraceEnabled()) logger.trace("Adding state {}: {}" , key, value.value); - reqdEndpointState.addApplicationState(key, value); + + states.put(key, value); } } + reqdEndpointState.addApplicationStates(states); } return reqdEndpointState; } @@ -1147,19 +1153,13 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean localState.setHeartBeatState(remoteState.getHeartBeatState()); if (logger.isTraceEnabled()) logger.trace("Updating heartbeat state version to {} from {} for {} ...", localState.getHeartBeatState().getHeartBeatVersion(), oldVersion, addr); - // we need to make two loops here, one to apply, then another to notify, this way all states in an update are present and current when the notifications are received - for (Entry remoteEntry : remoteState.getApplicationStateMap().entrySet()) - { - ApplicationState remoteKey = remoteEntry.getKey(); - VersionedValue remoteValue = remoteEntry.getValue(); - assert remoteState.getHeartBeatState().getGeneration() == localState.getHeartBeatState().getGeneration(); - localState.addApplicationState(remoteKey, remoteValue); - } - for (Entry remoteEntry : remoteState.getApplicationStateMap().entrySet()) - { + Set> remoteStates = remoteState.states(); + assert remoteState.getHeartBeatState().getGeneration() == localState.getHeartBeatState().getGeneration(); + localState.addApplicationStates(remoteStates); + + for (Entry remoteEntry : remoteStates) doOnChangeNotifications(addr, remoteEntry.getKey(), remoteEntry.getValue()); - } } // notify that a local application state is going to change (doesn't get triggered for remote changes) @@ -1273,7 +1273,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean public void start(int generationNumber) { - start(generationNumber, new HashMap()); + start(generationNumber, new EnumMap(ApplicationState.class)); } /** @@ -1285,8 +1285,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean /* initialize the heartbeat state for this localEndpoint */ maybeInitializeLocalState(generationNbr); EndpointState localState = endpointStateMap.get(FBUtilities.getBroadcastAddress()); - for (Map.Entry entry : preloadLocalStates.entrySet()) - localState.addApplicationState(entry.getKey(), entry.getValue()); + localState.addApplicationStates(preloadLocalStates); //notify snitches that Gossiper is about to start DatabaseDescriptor.getEndpointSnitch().gossiperStarting(); @@ -1475,8 +1474,10 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean EndpointState localState = oldState == null ? newState : oldState; // always add the version state - localState.addApplicationState(ApplicationState.NET_VERSION, StorageService.instance.valueFactory.networkVersion()); - localState.addApplicationState(ApplicationState.HOST_ID, StorageService.instance.valueFactory.hostId(uuid)); + Map states = new EnumMap<>(ApplicationState.class); + states.put(ApplicationState.NET_VERSION, StorageService.instance.valueFactory.networkVersion()); + states.put(ApplicationState.HOST_ID, StorageService.instance.valueFactory.hostId(uuid)); + localState.addApplicationStates(states); } @VisibleForTesting diff --git a/src/java/org/apache/cassandra/gms/VersionedValue.java b/src/java/org/apache/cassandra/gms/VersionedValue.java index a142f41b98..3ea7bb494f 100644 --- a/src/java/org/apache/cassandra/gms/VersionedValue.java +++ b/src/java/org/apache/cassandra/gms/VersionedValue.java @@ -109,6 +109,11 @@ public class VersionedValue implements Comparable return "Value(" + value + "," + version + ")"; } + public byte[] toBytes() + { + return value.getBytes(ISO_8859_1); + } + private static String versionString(String... args) { return StringUtils.join(args, VersionedValue.DELIMITER); diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index ad209fc46e..3ea261e276 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -31,6 +31,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.EnumMap; import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; @@ -521,9 +522,10 @@ public class StorageService extends NotificationBroadcasterSupport implements IE hostId = Gossiper.instance.getHostId(DatabaseDescriptor.getReplaceAddress()); try { - if (Gossiper.instance.getEndpointStateForEndpoint(DatabaseDescriptor.getReplaceAddress()).getApplicationState(ApplicationState.TOKENS) == null) + VersionedValue tokensVersionedValue = Gossiper.instance.getEndpointStateForEndpoint(DatabaseDescriptor.getReplaceAddress()).getApplicationState(ApplicationState.TOKENS); + if (tokensVersionedValue == null) throw new RuntimeException("Could not find tokens for " + DatabaseDescriptor.getReplaceAddress() + " to replace"); - Collection tokens = TokenSerializer.deserialize(getPartitioner(), new DataInputStream(new ByteArrayInputStream(getApplicationStateValue(DatabaseDescriptor.getReplaceAddress(), ApplicationState.TOKENS)))); + Collection tokens = TokenSerializer.deserialize(getPartitioner(), new DataInputStream(new ByteArrayInputStream(tokensVersionedValue.toBytes()))); SystemKeyspace.setLocalHostId(hostId); // use the replacee's host Id as our own so we receive hints, etc Gossiper.instance.resetEndpointStateMap(); // clean up since we have what we need @@ -740,7 +742,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE { if (!joined) { - Map appStates = new HashMap<>(); + Map appStates = new EnumMap<>(ApplicationState.class); if (replacing && !(Boolean.parseBoolean(System.getProperty("cassandra.join_ring", "true")))) throw new ConfigurationException("Cannot set both join_ring=false and attempt to replace a node"); @@ -1655,8 +1657,10 @@ public class StorageService extends NotificationBroadcasterSupport implements IE handleStateBootstrap(endpoint); break; case VersionedValue.STATUS_NORMAL: + handleStateNormal(endpoint, VersionedValue.STATUS_NORMAL); + break; case VersionedValue.SHUTDOWN: - handleStateNormal(endpoint); + handleStateNormal(endpoint, VersionedValue.SHUTDOWN); break; case VersionedValue.REMOVING_TOKEN: case VersionedValue.REMOVED_TOKEN: @@ -1738,7 +1742,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE private void updatePeerInfo(InetAddress endpoint) { EndpointState epState = Gossiper.instance.getEndpointStateForEndpoint(endpoint); - for (Map.Entry entry : epState.getApplicationStateMap().entrySet()) + for (Map.Entry entry : epState.states()) { switch (entry.getKey()) { @@ -1771,12 +1775,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE } } - private byte[] getApplicationStateValue(InetAddress endpoint, ApplicationState appstate) - { - String vvalue = Gossiper.instance.getEndpointStateForEndpoint(endpoint).getApplicationState(appstate).value; - return vvalue.getBytes(ISO_8859_1); - } - private void notifyRpcChange(InetAddress endpoint, boolean ready) { if (ready) @@ -1846,7 +1844,15 @@ public class StorageService extends NotificationBroadcasterSupport implements IE { try { - return TokenSerializer.deserialize(getPartitioner(), new DataInputStream(new ByteArrayInputStream(getApplicationStateValue(endpoint, ApplicationState.TOKENS)))); + EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(endpoint); + if (state == null) + return Collections.emptyList(); + + VersionedValue versionedValue = state.getApplicationState(ApplicationState.TOKENS); + if (versionedValue == null) + return Collections.emptyList(); + + return TokenSerializer.deserialize(getPartitioner(), new DataInputStream(new ByteArrayInputStream(versionedValue.toBytes()))); } catch (IOException e) { @@ -1895,22 +1901,23 @@ public class StorageService extends NotificationBroadcasterSupport implements IE * * @param endpoint node */ - private void handleStateNormal(final InetAddress endpoint) + private void handleStateNormal(final InetAddress endpoint, final String status) { - Collection tokens; - - tokens = getTokensFor(endpoint); - + Collection tokens = getTokensFor(endpoint); Set tokensToUpdateInMetadata = new HashSet<>(); Set tokensToUpdateInSystemKeyspace = new HashSet<>(); Set endpointsToRemove = new HashSet<>(); - if (logger.isDebugEnabled()) - logger.debug("Node {} state normal, token {}", endpoint, tokens); + logger.debug("Node {} state {}, token {}", endpoint, status, tokens); if (tokenMetadata.isMember(endpoint)) - logger.info("Node {} state jump to normal", endpoint); + logger.info("Node {} state jump to {}", endpoint, status); + + if (tokens.isEmpty() && status.equals(VersionedValue.STATUS_NORMAL)) + logger.error("Node {} is in state normal but it has no tokens, state: {}", + endpoint, + Gossiper.instance.getEndpointStateForEndpoint(endpoint)); updatePeerInfo(endpoint); // Order Matters, TM.updateHostID() should be called before TM.updateNormalToken(), (see CASSANDRA-4300). @@ -2021,8 +2028,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE */ private void handleStateLeaving(InetAddress endpoint) { - Collection tokens; - tokens = getTokensFor(endpoint); + Collection tokens = getTokensFor(endpoint); if (logger.isDebugEnabled()) logger.debug("Node {} state leaving, tokens {}", endpoint, tokens); @@ -2056,16 +2062,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE private void handleStateLeft(InetAddress endpoint, String[] pieces) { assert pieces.length >= 2; - Collection tokens = null; - try - { - tokens = getTokensFor(endpoint); - } - catch (Throwable th) - { - JVMStabilityInspector.inspectThrowable(th); - logger.warn("Unable to calculate tokens for {}.", endpoint); - } + Collection tokens = getTokensFor(endpoint); if (logger.isDebugEnabled()) logger.debug("Node {} state left, tokens {}", endpoint, tokens); @@ -2154,7 +2151,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE HintedHandOffManager.instance.deleteHintsForEndpoint(endpoint); removeEndpoint(endpoint); tokenMetadata.removeEndpoint(endpoint); - if (tokens != null) + if (!tokens.isEmpty()) tokenMetadata.removeBootstrapTokens(tokens); notifyLeft(endpoint); @@ -2358,7 +2355,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE public void onJoin(InetAddress endpoint, EndpointState epState) { - for (Map.Entry entry : epState.getApplicationStateMap().entrySet()) + for (Map.Entry entry : epState.states()) { onChange(endpoint, entry.getKey(), entry.getValue()); } diff --git a/test/unit/org/apache/cassandra/gms/EndpointStateTest.java b/test/unit/org/apache/cassandra/gms/EndpointStateTest.java new file mode 100644 index 0000000000..b06c435105 --- /dev/null +++ b/test/unit/org/apache/cassandra/gms/EndpointStateTest.java @@ -0,0 +1,159 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.gms; + +import java.util.Collections; +import java.util.EnumMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Test; + +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.dht.Token; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +public class EndpointStateTest +{ + public volatile VersionedValue.VersionedValueFactory valueFactory = + new VersionedValue.VersionedValueFactory(DatabaseDescriptor.getPartitioner()); + + @Test + public void testMultiThreadedReadConsistency() throws InterruptedException + { + for (int i = 0; i < 500; i++) + innerTestMultiThreadedReadConsistency(); + } + + /** + * Test that a thread reading values whilst they are updated by another thread will + * not see an entry unless it sees the entry previously added as well, even though + * we are accessing the map via an iterator backed by the underlying map. This + * works because EndpointState copies the map each time values are added. + */ + private void innerTestMultiThreadedReadConsistency() throws InterruptedException + { + final Token token = DatabaseDescriptor.getPartitioner().getRandomToken(); + final List tokens = Collections.singletonList(token); + final HeartBeatState hb = new HeartBeatState(0); + final EndpointState state = new EndpointState(hb); + final AtomicInteger numFailures = new AtomicInteger(); + + Thread t1 = new Thread(new Runnable() + { + public void run() + { + state.addApplicationState(ApplicationState.TOKENS, valueFactory.tokens(tokens)); + state.addApplicationState(ApplicationState.STATUS, valueFactory.normal(tokens)); + } + }); + + Thread t2 = new Thread(new Runnable() + { + public void run() + { + for (int i = 0; i < 50; i++) + { + Map values = new EnumMap<>(ApplicationState.class); + for (Map.Entry entry : state.states()) + values.put(entry.getKey(), entry.getValue()); + + if (values.containsKey(ApplicationState.STATUS) && !values.containsKey(ApplicationState.TOKENS)) + { + numFailures.incrementAndGet(); + System.out.println(String.format("Failed: %s", values)); + } + } + } + }); + + t1.start(); + t2.start(); + + t1.join(); + t2.join(); + + assertTrue(numFailures.get() == 0); + } + + @Test + public void testMultiThreadWriteConsistency() throws InterruptedException + { + for (int i = 0; i < 500; i++) + innerTestMultiThreadWriteConsistency(); + } + + /** + * Test that two threads can update the state map concurrently. + */ + private void innerTestMultiThreadWriteConsistency() throws InterruptedException + { + final Token token = DatabaseDescriptor.getPartitioner().getRandomToken(); + final List tokens = Collections.singletonList(token); + final String ip = "127.0.0.1"; + final UUID hostId = UUID.randomUUID(); + final HeartBeatState hb = new HeartBeatState(0); + final EndpointState state = new EndpointState(hb); + + Thread t1 = new Thread(new Runnable() + { + public void run() + { + Map states = new EnumMap<>(ApplicationState.class); + states.put(ApplicationState.TOKENS, valueFactory.tokens(tokens)); + states.put(ApplicationState.STATUS, valueFactory.normal(tokens)); + state.addApplicationStates(states); + } + }); + + Thread t2 = new Thread(new Runnable() + { + public void run() + { + Map states = new EnumMap<>(ApplicationState.class); + states.put(ApplicationState.INTERNAL_IP, valueFactory.internalIP(ip)); + states.put(ApplicationState.HOST_ID, valueFactory.hostId(hostId)); + state.addApplicationStates(states); + } + }); + + t1.start(); + t2.start(); + + t1.join(); + t2.join(); + + Set> states = state.states(); + assertEquals(4, states.size()); + + Map values = new EnumMap<>(ApplicationState.class); + for (Map.Entry entry : states) + values.put(entry.getKey(), entry.getValue()); + + assertTrue(values.containsKey(ApplicationState.STATUS)); + assertTrue(values.containsKey(ApplicationState.TOKENS)); + assertTrue(values.containsKey(ApplicationState.INTERNAL_IP)); + assertTrue(values.containsKey(ApplicationState.HOST_ID)); + } +} diff --git a/test/unit/org/apache/cassandra/locator/CloudstackSnitchTest.java b/test/unit/org/apache/cassandra/locator/CloudstackSnitchTest.java index d9a4ef1d0c..90e63e0caf 100644 --- a/test/unit/org/apache/cassandra/locator/CloudstackSnitchTest.java +++ b/test/unit/org/apache/cassandra/locator/CloudstackSnitchTest.java @@ -19,6 +19,7 @@ package org.apache.cassandra.locator; import java.io.IOException; import java.net.InetAddress; +import java.util.EnumMap; import java.util.Map; import org.junit.AfterClass; @@ -77,9 +78,10 @@ public class CloudstackSnitchTest InetAddress nonlocal = InetAddress.getByName("127.0.0.7"); Gossiper.instance.addSavedEndpoint(nonlocal); - Map stateMap = Gossiper.instance.getEndpointStateForEndpoint(nonlocal).getApplicationStateMap(); + Map stateMap = new EnumMap<>(ApplicationState.class); stateMap.put(ApplicationState.DC, StorageService.instance.valueFactory.datacenter("ch-zrh")); stateMap.put(ApplicationState.RACK, StorageService.instance.valueFactory.rack("2")); + Gossiper.instance.getEndpointStateForEndpoint(nonlocal).addApplicationStates(stateMap); assertEquals("ch-zrh", snitch.getDatacenter(nonlocal)); assertEquals("2", snitch.getRack(nonlocal)); diff --git a/test/unit/org/apache/cassandra/locator/EC2SnitchTest.java b/test/unit/org/apache/cassandra/locator/EC2SnitchTest.java index 6015adf1c9..56bbb775e5 100644 --- a/test/unit/org/apache/cassandra/locator/EC2SnitchTest.java +++ b/test/unit/org/apache/cassandra/locator/EC2SnitchTest.java @@ -24,6 +24,7 @@ package org.apache.cassandra.locator; import java.io.IOException; import java.net.InetAddress; import java.net.UnknownHostException; +import java.util.EnumMap; import java.util.Map; import org.junit.AfterClass; @@ -79,9 +80,10 @@ public class EC2SnitchTest InetAddress nonlocal = InetAddress.getByName("127.0.0.7"); Gossiper.instance.addSavedEndpoint(nonlocal); - Map stateMap = Gossiper.instance.getEndpointStateForEndpoint(nonlocal).getApplicationStateMap(); + Map stateMap = new EnumMap<>(ApplicationState.class); stateMap.put(ApplicationState.DC, StorageService.instance.valueFactory.datacenter("us-west")); stateMap.put(ApplicationState.RACK, StorageService.instance.valueFactory.datacenter("1a")); + Gossiper.instance.getEndpointStateForEndpoint(nonlocal).addApplicationStates(stateMap); assertEquals("us-west", snitch.getDatacenter(nonlocal)); assertEquals("1a", snitch.getRack(nonlocal)); diff --git a/test/unit/org/apache/cassandra/locator/GoogleCloudSnitchTest.java b/test/unit/org/apache/cassandra/locator/GoogleCloudSnitchTest.java index 54ea722b2f..1521454ece 100644 --- a/test/unit/org/apache/cassandra/locator/GoogleCloudSnitchTest.java +++ b/test/unit/org/apache/cassandra/locator/GoogleCloudSnitchTest.java @@ -23,6 +23,7 @@ package org.apache.cassandra.locator; import java.io.IOException; import java.net.InetAddress; +import java.util.EnumMap; import java.util.Map; import org.junit.AfterClass; @@ -75,9 +76,10 @@ public class GoogleCloudSnitchTest InetAddress nonlocal = InetAddress.getByName("127.0.0.7"); Gossiper.instance.addSavedEndpoint(nonlocal); - Map stateMap = Gossiper.instance.getEndpointStateForEndpoint(nonlocal).getApplicationStateMap(); + Map stateMap = new EnumMap<>(ApplicationState.class); stateMap.put(ApplicationState.DC, StorageService.instance.valueFactory.datacenter("europe-west1")); stateMap.put(ApplicationState.RACK, StorageService.instance.valueFactory.datacenter("a")); + Gossiper.instance.getEndpointStateForEndpoint(nonlocal).addApplicationStates(stateMap); assertEquals("europe-west1", snitch.getDatacenter(nonlocal)); assertEquals("a", snitch.getRack(nonlocal));