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

Apply changes to gossip state for the local node synchronously, remote peers asynchronously 

Patch by Sam Tunnicliffe; reviewed by Alex Petrov for CASSANDRA-21239

Co-authored-by: Alex Petrov <oleksandr.petrov@gmail.com>
This commit is contained in:
Sam Tunnicliffe 2026-04-20 16:27:16 +01:00
parent 3f1593064c
commit 53f188293d
14 changed files with 229 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
6.0-alpha2
* Synchronously publish changes to local gossip state following metadata updates (CASSANDRA-21239)
* Change default for cassandra.set_sep_thread_name to false to reduce CPU usage (CASSANDRA-21089)
* Avoid permission checks for masked columns when the table doesn't have any (CASSANDRA-21299)
* Reduce allocations and array copies due to buffer resizing in LocalDataResponse during row serialization (CASSANDRA-21285)

1
src/java/org/apache/cassandra/config/Config.java
View File

@ -1503,6 +1503,7 @@ public class Config
public volatile DurationSpec.LongMillisecondsBound progress_barrier_backoff = new DurationSpec.LongMillisecondsBound("1000ms");
public volatile DurationSpec.LongSecondsBound discovery_timeout = new DurationSpec.LongSecondsBound("30s");
public boolean unsafe_tcm_mode = false;
public boolean legacy_state_listener_sync_local_updates = true;
public enum TriggersPolicy
{

20
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -6198,6 +6198,26 @@ public class DatabaseDescriptor
return conf.unsafe_tcm_mode;
}
public static boolean getLegacyStateListenerSyncLocalUpdates()
{
return conf.legacy_state_listener_sync_local_updates;
}
public static void setLegacyStateListenerSyncLocalUpdates(boolean sync)
{
if (sync != conf.legacy_state_listener_sync_local_updates)
{
logger.info("Changing processing mode of state updates to the local node in LegacyStateListener from {} to {}",
sync ? "async" : "sync", sync ? "sync" : "async");
conf.legacy_state_listener_sync_local_updates = sync;
}
else
{
logger.info("Not changing processing mode of state updates to the local node in LegacyStateListener, already set to {}",
sync ? "sync" : "async");
}
}
public static int getSaiSSTableIndexesPerQueryWarnThreshold()
{
return conf.sai_sstable_indexes_per_query_warn_threshold;

2
src/java/org/apache/cassandra/service/accord/AccordService.java
View File

@ -161,7 +161,7 @@ import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class AccordService implements IAccordService, Shutdownable
{
public static class MetadataChangeListener implements ChangeListener.Async
public static class MetadataChangeListener implements ChangeListener
{
// Listener is initialized before Accord is initialized
public static MetadataChangeListener instance = new MetadataChangeListener();

12
src/java/org/apache/cassandra/tcm/CMSOperations.java
View File

@ -280,6 +280,18 @@ public class CMSOperations implements CMSOperationsMBean
return convertToStringValues(log);
}
@Override
public boolean getLegacyStateListenerSyncLocalUpdates()
{
return DatabaseDescriptor.getLegacyStateListenerSyncLocalUpdates();
}
@Override
public void setLegacyStateListenerSyncLocalUpdates(boolean sync)
{
DatabaseDescriptor.setLegacyStateListenerSyncLocalUpdates(sync);
}
private Map<Long, Map<String, String>> convertToStringValues(Map<Long, Map<String, Object>> log)
{
Map<Long, Map<String, String>> res = new LinkedHashMap<>();

3
src/java/org/apache/cassandra/tcm/CMSOperationsMBean.java
View File

@ -50,4 +50,7 @@ public interface CMSOperationsMBean
public Map<Long, Map<String, String>> dumpLog(long startEpoch, long endEpoch);
public void resumeDropAccordTable(String tableId);
public boolean getLegacyStateListenerSyncLocalUpdates();
public void setLegacyStateListenerSyncLocalUpdates(boolean sync);
}

7
src/java/org/apache/cassandra/tcm/compatibility/GossipHelper.java
View File

@ -105,9 +105,12 @@ public class GossipHelper
Gossiper.runInGossipStageBlocking(() -> Gossiper.instance.removeEndpoint(addr));
}
public static void evictFromMembership(InetAddressAndPort endpoint)
public static void removeAndEvict(InetAddressAndPort endpoint)
{
Gossiper.runInGossipStageBlocking(() -> Gossiper.instance.evictFromMembership(endpoint));
Gossiper.runInGossipStageBlocking(() -> {
Gossiper.instance.removeEndpoint(endpoint);
Gossiper.instance.evictFromMembership(endpoint);
});
}
public static VersionedValue nodeStateToStatus(NodeId nodeId,

2
src/java/org/apache/cassandra/tcm/listeners/ChangeListener.java
View File

@ -37,6 +37,4 @@ public interface ChangeListener
*/
default void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot) {}
interface Async extends ChangeListener {}
}

168
src/java/org/apache/cassandra/tcm/listeners/LegacyStateListener.java
View File

@ -29,6 +29,8 @@ import com.google.common.collect.Sets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.db.virtual.PeersTable;
@ -54,7 +56,7 @@ import static org.apache.cassandra.tcm.membership.NodeState.LEFT;
import static org.apache.cassandra.tcm.membership.NodeState.MOVING;
import static org.apache.cassandra.tcm.membership.NodeState.REGISTERED;
public class LegacyStateListener implements ChangeListener.Async
public class LegacyStateListener implements ChangeListener
{
private static final Logger logger = LoggerFactory.getLogger(LegacyStateListener.class);
@ -75,52 +77,97 @@ public class LegacyStateListener implements ChangeListener.Async
changed.add(node);
}
for (InetAddressAndPort remove : removedAddr)
// next.myNodeId() can be null during replay (before we have registered) but if it is present and
// there is a relevant change to the state of the local node, process that synchronously.
if (next.myNodeId() != null && changed.contains(next.myNodeId()))
{
GossipHelper.removeFromGossip(remove);
GossipHelper.evictFromMembership(remove);
PeersTable.removeFromSystemPeersTables(remove);
// Default is to process updates for the local node synchronously, overridable via config/hotprop
if (DatabaseDescriptor.getLegacyStateListenerSyncLocalUpdates())
processChangesToLocalState(prev, next, next.myNodeId());
else
ScheduledExecutors.optionalTasks.submit(() -> processChangesToLocalState(prev, next, next.myNodeId()));
changed.remove(next.myNodeId());
}
// Schedule async processing of changes to peers and removing unregistered nodes (potentially including the
// local node).
ScheduledExecutors.optionalTasks.submit(() -> {
processRemovedNodes(removedAddr);
processChangesToRemotePeers(prev, next, changed);
});
}
private void processChangesToLocalState(ClusterMetadata prev, ClusterMetadata next, NodeId localId)
{
logger.info("Processing changes to local node state {} for epoch {}->{}", localId, prev.epoch.getEpoch(), next.epoch.getEpoch());
Collection<Token> tokensForGossip = next.tokenMap.tokens(localId);
NodeState state = next.directory.peerState(localId);
switch (state)
{
case BOOTSTRAPPING:
case BOOT_REPLACING:
// For compatibility with clients, ensure we set TOKENS for bootstrapping nodes in gossip.
// As these are not yet added to the token map they must be extracted from the in progress sequence.
tokensForGossip = GossipHelper.getTokensFromOperation(localId, next);
if (state == BOOTSTRAPPING && prev.directory.peerState(localId) != BOOTSTRAPPING)
{
// legacy log messages for tests
logger.info("JOINING: Starting to bootstrap");
logger.info("JOINING: calculation complete, ready to bootstrap");
}
break;
case JOINED:
tokensForGossip = next.tokenMap.tokens(localId);
SystemKeyspace.updateTokens(next.directory.endpoint(localId), tokensForGossip);
Set<String> userKeyspaces = Schema.instance.getUserKeyspaces().names();
StreamSupport.stream(ColumnFamilyStore.all().spliterator(), false)
.filter(cfs -> userKeyspaces.contains(cfs.keyspace.getName()))
.forEach(cfs -> cfs.indexManager.executePreJoinTasksBlocking(true));
NodeState previousState = prev.directory.peerState(localId);
if (previousState == MOVING)
{
logger.info("Node {} state jump to NORMAL", next.directory.endpoint(localId));
}
else if (previousState == BOOT_REPLACING)
{
// legacy log message for compatibility (& tests)
MultiStepOperation<?> sequence = prev.inProgressSequences.get(localId);
if (sequence != null && sequence.kind() == MultiStepOperation.Kind.REPLACE)
{
logCompletedReplacement(prev.directory, (BootstrapAndReplace) sequence);
tokensForGossip = GossipHelper.getTokensFromOperation(sequence);
}
}
break;
case MOVING:
logger.debug("Node {} state MOVING, tokens {}", next.directory.endpoint(localId), prev.tokenMap.tokens(localId));
tokensForGossip = next.tokenMap.tokens(localId);
break;
case LEFT:
tokensForGossip = prev.tokenMap.tokens(localId);
break;
}
// Maybe initialise local epstate whatever the node state because we could be processing after a
// replay and so may have not seen any previous local states, making this the first mutation of gossip
// state for the local node.
Gossiper.instance.maybeInitializeLocalState(SystemKeyspace.incrementAndGetGeneration());
Gossiper.instance.addLocalApplicationState(SCHEMA, StorageService.instance.valueFactory.schema(next.schema.getVersion()));
// Pull node properties from cluster metadata into gossip, except if the node is only in the REGISTERED state
// as that has no equivalent gossip STATUS
if (state != REGISTERED)
Gossiper.instance.mergeNodeToGossip(localId, next, tokensForGossip);
// if the local node's location has changed, update system.local.
if (!next.directory.location(localId).equals(prev.directory.location(localId)))
SystemKeyspace.updateLocation(next.directory.location(localId));
}
private void processChangesToRemotePeers(ClusterMetadata prev, ClusterMetadata next, Set<NodeId> changed)
{
for (NodeId change : changed)
{
// next.myNodeId() can be null during replay (before we have registered)
if (next.myNodeId() != null && next.myNodeId().equals(change))
{
switch (next.directory.peerState(change))
{
case BOOTSTRAPPING:
if (prev.directory.peerState(change) != BOOTSTRAPPING)
{
// legacy log messages for tests
logger.info("JOINING: Starting to bootstrap");
logger.info("JOINING: calculation complete, ready to bootstrap");
}
break;
case BOOT_REPLACING:
case REGISTERED:
break;
case JOINED:
SystemKeyspace.updateTokens(next.directory.endpoint(change), next.tokenMap.tokens(change));
// needed if we miss the REGISTERED above; Does nothing if we are already in epStateMap:
Gossiper.instance.maybeInitializeLocalState(SystemKeyspace.incrementAndGetGeneration());
StreamSupport.stream(ColumnFamilyStore.all().spliterator(), false)
.filter(cfs -> Schema.instance.getUserKeyspaces().names().contains(cfs.keyspace.getName()))
.forEach(cfs -> cfs.indexManager.executePreJoinTasksBlocking(true));
if (prev.directory.peerState(change) == MOVING)
logger.info("Node {} state jump to NORMAL", next.directory.endpoint(change));
break;
}
// Maybe intitialise local epstate whatever the node state because we could be processing after a
// replay and so may have not seen any previous local states, making this the first mutation of gossip
// state for the local node.
Gossiper.instance.maybeInitializeLocalState(SystemKeyspace.incrementAndGetGeneration());
Gossiper.instance.addLocalApplicationState(SCHEMA, StorageService.instance.valueFactory.schema(next.schema.getVersion()));
// if the local node's location has changed, update system.local.
if (!next.directory.location(change).equals(prev.directory.location(change)))
SystemKeyspace.updateLocation(next.directory.location(change));
}
logger.info("Processing changes to peer {} for epoch {}->{}", change, prev.epoch.getEpoch(), next.epoch.getEpoch());
if (next.directory.peerState(change) == REGISTERED)
{
// Re-establish any connections made prior to this node registering
@ -155,21 +202,11 @@ public class LegacyStateListener implements ChangeListener.Async
}
else if (prev.directory.peerState(change) == BOOT_REPLACING)
{
// legacy log message for compatibility (& tests)
MultiStepOperation<?> sequence = prev.inProgressSequences.get(change);
if (sequence != null && sequence.kind() == MultiStepOperation.Kind.REPLACE)
{
BootstrapAndReplace replace = (BootstrapAndReplace) sequence;
InetAddressAndPort replaced = prev.directory.endpoint(replace.startReplace.replaced());
InetAddressAndPort replacement = prev.directory.endpoint(change);
Collection<Token> tokens = GossipHelper.getTokensFromOperation(replace);
logger.info("Node {} will complete replacement of {} for tokens {}", replacement, replaced, tokens);
if (!replacement.equals(replaced))
{
for (Token token : tokens)
logger.warn("Token {} changing ownership from {} to {}", token, replaced, replacement);
}
Gossiper.instance.mergeNodeToGossip(change, next, tokens);
logCompletedReplacement(prev.directory, (BootstrapAndReplace) sequence);
Gossiper.instance.mergeNodeToGossip(change, next, GossipHelper.getTokensFromOperation(sequence));
PeersTable.updateLegacyPeerTable(change, prev, next);
}
}
@ -181,6 +218,29 @@ public class LegacyStateListener implements ChangeListener.Async
}
}
private void processRemovedNodes(Set<InetAddressAndPort> removed)
{
for (InetAddressAndPort remove : removed)
{
GossipHelper.removeAndEvict(remove);
PeersTable.removeFromSystemPeersTables(remove);
}
}
private void logCompletedReplacement(Directory directory, BootstrapAndReplace sequence)
{
// legacy log message for compatibility (& tests)
InetAddressAndPort replaced = directory.endpoint(sequence.startReplace.replaced());
InetAddressAndPort replacement = directory.endpoint(sequence.startReplace.replacement());
Collection<Token> tokens = GossipHelper.getTokensFromOperation(sequence);
logger.info("Node {} will complete replacement of {} for tokens {}", replacement, replaced, tokens);
if (!replacement.equals(replaced))
{
for (Token token : tokens)
logger.warn("Token {} changing ownership from {} to {}", token, replaced, replacement);
}
}
private boolean directoryEntryChangedFor(NodeId nodeId, Directory prev, Directory next)
{
return prev.peerState(nodeId) != next.peerState(nodeId) ||

33
src/java/org/apache/cassandra/tcm/listeners/UpgradeMigrationListener.java
View File

@ -24,18 +24,39 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.utils.CassandraVersion;
public class UpgradeMigrationListener implements ChangeListener.Async
/**
* For handling changes in Cassandra version.
* One use case is to react to the initial migration from Gossip based metadata in Cassandra 5.0 and earlier. When
* a node first transitions to using ClusterMetadataService, this listener will update its gossip state with the new
* nodeId based hostId and ensure that is propagated.
*
* Another use is for evolving distributed system tables, this listener can identify when a new Cassandra version has
* been deployed across the cluster and provides a hook to take actions such as creating new internal tables etc.
*/
public class UpgradeMigrationListener implements ChangeListener
{
private static final Logger logger = LoggerFactory.getLogger(UpgradeMigrationListener.class);
public void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot)
{
if (!prev.epoch.equals(Epoch.UPGRADE_GOSSIP))
if (prev.epoch.equals(Epoch.UPGRADE_GOSSIP))
{
logger.info("Detected upgrade from gossip mode, updating my host id in gossip to {}", next.myNodeId());
Gossiper.instance.mergeNodeToGossip(next.myNodeId(), next);
if (Gossiper.instance.getQuarantineDisabled())
Gossiper.instance.clearQuarantinedEndpoints();
return;
}
logger.info("Detected upgrade from gossip mode, updating my host id in gossip to {}", next.myNodeId());
Gossiper.instance.mergeNodeToGossip(next.myNodeId(), next);
if (Gossiper.instance.getQuarantineDisabled())
Gossiper.instance.clearQuarantinedEndpoints();
CassandraVersion prevMinVersion = prev.directory.clusterMinVersion.cassandraVersion;
CassandraVersion minVersion = next.directory.clusterMinVersion.cassandraVersion;
if (prevMinVersion.compareTo(minVersion) == 0 || (prev.epoch.is(Epoch.EMPTY) && fromSnapshot))
{
// nothing to do if the min version in the cluster has not changed
// likewise, we don't need to trigger if applying a snapshot to a previously empty cluster metadata for e.g.
// when replaying at startup
logger.debug("Cluster min version has not changed, nothing to do");
}
}
}

21
src/java/org/apache/cassandra/tcm/log/LocalLog.java
View File

@ -121,7 +121,6 @@ public abstract class LocalLog implements Closeable
private final Set<LogListener> listeners = new HashSet<>();
private final Set<ChangeListener> changeListeners = new HashSet<>();
private final Set<ChangeListener.Async> asyncChangeListeners = new HashSet<>();
private LogSpec()
{
@ -157,7 +156,7 @@ public abstract class LocalLog implements Closeable
public LogSpec withDefaultListeners(boolean withDefaultListeners)
{
if (withDefaultListeners &&
!(listeners.isEmpty() && changeListeners.isEmpty() && asyncChangeListeners.isEmpty()))
!(listeners.isEmpty() && changeListeners.isEmpty()))
{
throw new IllegalStateException("LogSpec can only require all listeners OR specific listeners");
}
@ -178,10 +177,7 @@ public abstract class LocalLog implements Closeable
{
if (defaultListeners)
throw new IllegalStateException("LogSpec can only require all listeners OR specific listeners");
if (listener instanceof ChangeListener.Async)
asyncChangeListeners.add((ChangeListener.Async) listener);
else
changeListeners.add(listener);
changeListeners.add(listener);
return this;
}
@ -257,7 +253,6 @@ public abstract class LocalLog implements Closeable
protected final LogStorage storage;
protected final Set<LogListener> listeners;
protected final Set<ChangeListener> changeListeners;
protected final Set<ChangeListener.Async> asyncChangeListeners;
protected final LogSpec spec;
// for testing - used to inject filters which cause entries to be dropped before appending
@ -277,7 +272,6 @@ public abstract class LocalLog implements Closeable
this.storage = logSpec.storage;
listeners = Sets.newConcurrentHashSet();
changeListeners = Sets.newConcurrentHashSet();
asyncChangeListeners = Sets.newConcurrentHashSet();
entryFilters = Lists.newCopyOnWriteArrayList();
}
@ -596,10 +590,7 @@ public abstract class LocalLog implements Closeable
public void addListener(ChangeListener listener)
{
if (listener instanceof ChangeListener.Async)
this.asyncChangeListeners.add((ChangeListener.Async) listener);
else
this.changeListeners.add(listener);
this.changeListeners.add(listener);
}
public void removeListener(ChangeListener listener)
@ -619,16 +610,12 @@ public abstract class LocalLog implements Closeable
{
for (ChangeListener listener : changeListeners)
listener.notifyPreCommit(before, after, fromSnapshot);
for (ChangeListener.Async listener : asyncChangeListeners)
ScheduledExecutors.optionalTasks.submit(() -> listener.notifyPreCommit(before, after, fromSnapshot));
}
private void notifyPostCommit(ClusterMetadata before, ClusterMetadata after, boolean fromSnapshot)
{
for (ChangeListener listener : changeListeners)
listener.notifyPostCommit(before, after, fromSnapshot);
for (ChangeListener.Async listener : asyncChangeListeners)
ScheduledExecutors.optionalTasks.submit(() -> listener.notifyPostCommit(before, after, fromSnapshot));
}
public void addFilter(Predicate<Entry> filter)
@ -680,7 +667,6 @@ public abstract class LocalLog implements Closeable
logger.info("Adding specified listeners to LocalLog");
spec.listeners.forEach(this::addListener);
spec.changeListeners.forEach(this::addListener);
spec.asyncChangeListeners.forEach(this::addListener);
}
logger.info("Notifying all registered listeners of both pre and post commit event");
@ -917,7 +903,6 @@ public abstract class LocalLog implements Closeable
{
listeners.clear();
changeListeners.clear();
asyncChangeListeners.clear();
addListener(snapshotListener());
addListener(new InitializationListener());

1
src/java/org/apache/cassandra/tcm/transformations/cms/PrepareCMSReconfiguration.java
View File

@ -81,6 +81,7 @@ public abstract class PrepareCMSReconfiguration implements Transformation
return Transformation.success(prev.transformer(), LockedRanges.AffectedRanges.EMPTY);
}
logger.info("Proposed CMS reconfiguration resulted in {}", diff);
LockedRanges.Key lockKey = LockedRanges.keyFor(prev.nextEpoch());
Set<NodeId> cms = prev.fullCMSMembers().stream().map(prev.directory::peerId).collect(Collectors.toSet());
Set<NodeId> tmp = new HashSet<>(cms);

27
test/distributed/org/apache/cassandra/distributed/test/log/InProgressSequenceCoordinationTest.java
View File

@ -22,7 +22,9 @@ import java.util.Random;
import java.util.concurrent.Callable;
import java.util.function.BiFunction;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.apache.cassandra.distributed.Cluster;
@ -53,6 +55,7 @@ import org.apache.cassandra.tcm.transformations.PrepareReplace;
import org.apache.cassandra.utils.concurrent.Condition;
import static org.apache.cassandra.config.CassandraRelevantProperties.REPLACE_ADDRESS_FIRST_BOOT;
import static org.apache.cassandra.config.CassandraRelevantProperties.TCM_SKIP_CMS_RECONFIGURATION_AFTER_TOPOLOGY_CHANGE;
import static org.apache.cassandra.distributed.Constants.KEY_DTEST_API_STARTUP_FAILURE_AS_SHUTDOWN;
import static org.apache.cassandra.distributed.Constants.KEY_DTEST_FULL_STARTUP;
import static org.apache.cassandra.distributed.shared.ClusterUtils.addInstance;
@ -65,6 +68,30 @@ import static org.apache.cassandra.tcm.sequences.SequenceState.halted;
public class InProgressSequenceCoordinationTest extends FuzzTestBase
{
private boolean skipReconfiguation;
@Before
public void setup()
{
// Skip the automatic, speculative CMS reconfiguration after join/replace operations.
// These can cause tests to run long or hang as peers start to shutdown while they are
// in flight. If the node initiating the reconfiguration sees the current CMS node as
// DOWN, a reconfiguration is triggered which cannot be be completed in a timely fashion.
// Previously, this was hidden because LegacyStateListener would queue up gossip tasks
// on a separate executor and the backlog of these prevented the failure detector from
// marking any nodes DOWN. Since making LegacyStateListener run synchronously on the
// log follower thread, the joining node does see the CMS node as DOWN and so tries to
// perform a reconfiguration.
skipReconfiguation = TCM_SKIP_CMS_RECONFIGURATION_AFTER_TOPOLOGY_CHANGE.getBoolean();
TCM_SKIP_CMS_RECONFIGURATION_AFTER_TOPOLOGY_CHANGE.setBoolean(true);
}
@After
public void tearDown()
{
TCM_SKIP_CMS_RECONFIGURATION_AFTER_TOPOLOGY_CHANGE.setBoolean(skipReconfiguation);
}
@Test
public void bootstrapProgressTest() throws Throwable
{

18
test/unit/org/apache/cassandra/service/JoinTokenRingTest.java
View File

@ -27,13 +27,17 @@ import org.junit.Test;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.gms.ApplicationState;
import org.apache.cassandra.gms.EndpointState;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.gms.VersionedValue;
import org.apache.cassandra.index.SecondaryIndexManager;
import org.apache.cassandra.index.StubIndex;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.FBUtilities;
@ -51,12 +55,18 @@ public class JoinTokenRingTest
@Test
public void testIndexPreJoinInvocation() throws IOException, ExecutionException, InterruptedException
{
ClusterMetadataTestHelper.addEndpoint(FBUtilities.getBroadcastAddressAndPort(),
ClusterMetadata.current().partitioner.getRandomToken());
ScheduledExecutors.optionalTasks.submit(() -> null).get(); // make sure the LegacyStateListener has finished executing
InetAddressAndPort local = FBUtilities.getBroadcastAddressAndPort();
ClusterMetadataTestHelper.addEndpoint(local, ClusterMetadata.current().partitioner.getRandomToken());
SecondaryIndexManager indexManager = ColumnFamilyStore.getIfExists("JoinTokenRingTestKeyspace7", "Indexed1").indexManager;
StubIndex stub = (StubIndex) indexManager.getIndexByName("Indexed1_value_index");
Assert.assertTrue(stub.preJoinInvocation);
EndpointState epState = Gossiper.instance.getEndpointStateForEndpoint(local);
Assert.assertNotNull(epState);
VersionedValue statusValue = epState.getApplicationState(ApplicationState.STATUS_WITH_PORT);
Assert.assertNotNull(statusValue);
Assert.assertTrue(statusValue.value.contains(VersionedValue.STATUS_NORMAL));
}