Skip to content

Commit 0c6380f

Browse files
beobalbeobal
committed
Merge branch 'cassandra-6.0' into trunk
2 parents c5d0129 + 53f1882 commit 0c6380f

14 files changed

Lines changed: 227 additions & 85 deletions

File tree

‎CHANGES.txt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33

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

‎src/java/org/apache/cassandra/config/Config.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1503,6 +1503,7 @@ public static void log(Config config)
15031503
public volatile DurationSpec.LongMillisecondsBound progress_barrier_backoff = new DurationSpec.LongMillisecondsBound("1000ms");
15041504
public volatile DurationSpec.LongSecondsBound discovery_timeout = new DurationSpec.LongSecondsBound("30s");
15051505
public boolean unsafe_tcm_mode = false;
1506+
public boolean legacy_state_listener_sync_local_updates = true;
15061507

15071508
public enum TriggersPolicy
15081509
{

‎src/java/org/apache/cassandra/config/DatabaseDescriptor.java‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6198,6 +6198,26 @@ public static boolean getUnsafeTCMMode()
61986198
return conf.unsafe_tcm_mode;
61996199
}
62006200

6201+
public static boolean getLegacyStateListenerSyncLocalUpdates()
6202+
{
6203+
return conf.legacy_state_listener_sync_local_updates;
6204+
}
6205+
6206+
public static void setLegacyStateListenerSyncLocalUpdates(boolean sync)
6207+
{
6208+
if (sync != conf.legacy_state_listener_sync_local_updates)
6209+
{
6210+
logger.info("Changing processing mode of state updates to the local node in LegacyStateListener from {} to {}",
6211+
sync ? "async" : "sync", sync ? "sync" : "async");
6212+
conf.legacy_state_listener_sync_local_updates = sync;
6213+
}
6214+
else
6215+
{
6216+
logger.info("Not changing processing mode of state updates to the local node in LegacyStateListener, already set to {}",
6217+
sync ? "sync" : "async");
6218+
}
6219+
}
6220+
62016221
public static int getSaiSSTableIndexesPerQueryWarnThreshold()
62026222
{
62036223
return conf.sai_sstable_indexes_per_query_warn_threshold;

‎src/java/org/apache/cassandra/service/accord/AccordService.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -161,7 +161,7 @@
161161

162162
public class AccordService implements IAccordService, Shutdownable
163163
{
164-
public static class MetadataChangeListener implements ChangeListener.Async
164+
public static class MetadataChangeListener implements ChangeListener
165165
{
166166
// Listener is initialized before Accord is initialized
167167
public static MetadataChangeListener instance = new MetadataChangeListener();

‎src/java/org/apache/cassandra/tcm/CMSOperations.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -280,6 +280,18 @@ public Map<Long, Map<String, String>> dumpLog(long startEpoch, long endEpoch)
280280
return convertToStringValues(log);
281281
}
282282

283+
@Override
284+
public boolean getLegacyStateListenerSyncLocalUpdates()
285+
{
286+
return DatabaseDescriptor.getLegacyStateListenerSyncLocalUpdates();
287+
}
288+
289+
@Override
290+
public void setLegacyStateListenerSyncLocalUpdates(boolean sync)
291+
{
292+
DatabaseDescriptor.setLegacyStateListenerSyncLocalUpdates(sync);
293+
}
294+
283295
private Map<Long, Map<String, String>> convertToStringValues(Map<Long, Map<String, Object>> log)
284296
{
285297
Map<Long, Map<String, String>> res = new LinkedHashMap<>();

‎src/java/org/apache/cassandra/tcm/CMSOperationsMBean.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,4 +50,7 @@ public interface CMSOperationsMBean
5050
public Map<Long, Map<String, String>> dumpLog(long startEpoch, long endEpoch);
5151

5252
public void resumeDropAccordTable(String tableId);
53+
54+
public boolean getLegacyStateListenerSyncLocalUpdates();
55+
public void setLegacyStateListenerSyncLocalUpdates(boolean sync);
5356
}

‎src/java/org/apache/cassandra/tcm/compatibility/GossipHelper.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -105,9 +105,12 @@ public static void removeFromGossip(InetAddressAndPort addr)
105105
Gossiper.runInGossipStageBlocking(() -> Gossiper.instance.removeEndpoint(addr));
106106
}
107107

108-
public static void evictFromMembership(InetAddressAndPort endpoint)
108+
public static void removeAndEvict(InetAddressAndPort endpoint)
109109
{
110-
Gossiper.runInGossipStageBlocking(() -> Gossiper.instance.evictFromMembership(endpoint));
110+
Gossiper.runInGossipStageBlocking(() -> {
111+
Gossiper.instance.removeEndpoint(endpoint);
112+
Gossiper.instance.evictFromMembership(endpoint);
113+
});
111114
}
112115

113116
public static VersionedValue nodeStateToStatus(NodeId nodeId,

‎src/java/org/apache/cassandra/tcm/listeners/ChangeListener.java‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,4 @@ default void notifyPreCommit(ClusterMetadata prev, ClusterMetadata next, boolean
3737
*/
3838
default void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot) {}
3939

40-
interface Async extends ChangeListener {}
41-
4240
}

‎src/java/org/apache/cassandra/tcm/listeners/LegacyStateListener.java‎

Lines changed: 112 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,8 @@
2929
import org.slf4j.Logger;
3030
import org.slf4j.LoggerFactory;
3131

32+
import org.apache.cassandra.concurrent.ScheduledExecutors;
33+
import org.apache.cassandra.config.DatabaseDescriptor;
3234
import org.apache.cassandra.db.ColumnFamilyStore;
3335
import org.apache.cassandra.db.SystemKeyspace;
3436
import org.apache.cassandra.db.virtual.PeersTable;
@@ -54,7 +56,7 @@
5456
import static org.apache.cassandra.tcm.membership.NodeState.MOVING;
5557
import static org.apache.cassandra.tcm.membership.NodeState.REGISTERED;
5658

57-
public class LegacyStateListener implements ChangeListener.Async
59+
public class LegacyStateListener implements ChangeListener
5860
{
5961
private static final Logger logger = LoggerFactory.getLogger(LegacyStateListener.class);
6062

@@ -75,52 +77,97 @@ public void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean
7577
changed.add(node);
7678
}
7779

78-
for (InetAddressAndPort remove : removedAddr)
80+
// next.myNodeId() can be null during replay (before we have registered) but if it is present and
81+
// there is a relevant change to the state of the local node, process that synchronously.
82+
if (next.myNodeId() != null && changed.contains(next.myNodeId()))
7983
{
80-
GossipHelper.removeFromGossip(remove);
81-
GossipHelper.evictFromMembership(remove);
82-
PeersTable.removeFromSystemPeersTables(remove);
84+
// Default is to process updates for the local node synchronously, overridable via config/hotprop
85+
if (DatabaseDescriptor.getLegacyStateListenerSyncLocalUpdates())
86+
processChangesToLocalState(prev, next, next.myNodeId());
87+
else
88+
ScheduledExecutors.optionalTasks.submit(() -> processChangesToLocalState(prev, next, next.myNodeId()));
89+
90+
changed.remove(next.myNodeId());
8391
}
8492

85-
for (NodeId change : changed)
93+
// Schedule async processing of changes to peers and removing unregistered nodes (potentially including the
94+
// local node).
95+
ScheduledExecutors.optionalTasks.submit(() -> {
96+
processRemovedNodes(removedAddr);
97+
processChangesToRemotePeers(prev, next, changed);
98+
});
99+
}
100+
101+
private void processChangesToLocalState(ClusterMetadata prev, ClusterMetadata next, NodeId localId)
102+
{
103+
logger.info("Processing changes to local node state {} for epoch {}->{}", localId, prev.epoch.getEpoch(), next.epoch.getEpoch());
104+
Collection<Token> tokensForGossip = next.tokenMap.tokens(localId);
105+
NodeState state = next.directory.peerState(localId);
106+
switch (state)
86107
{
87-
// next.myNodeId() can be null during replay (before we have registered)
88-
if (next.myNodeId() != null && next.myNodeId().equals(change))
89-
{
90-
switch (next.directory.peerState(change))
108+
case BOOTSTRAPPING:
109+
case BOOT_REPLACING:
110+
// For compatibility with clients, ensure we set TOKENS for bootstrapping nodes in gossip.
111+
// As these are not yet added to the token map they must be extracted from the in progress sequence.
112+
tokensForGossip = GossipHelper.getTokensFromOperation(localId, next);
113+
if (state == BOOTSTRAPPING && prev.directory.peerState(localId) != BOOTSTRAPPING)
91114
{
92-
case BOOTSTRAPPING:
93-
if (prev.directory.peerState(change) != BOOTSTRAPPING)
94-
{
95-
// legacy log messages for tests
96-
logger.info("JOINING: Starting to bootstrap");
97-
logger.info("JOINING: calculation complete, ready to bootstrap");
98-
}
99-
break;
100-
case BOOT_REPLACING:
101-
case REGISTERED:
102-
break;
103-
case JOINED:
104-
SystemKeyspace.updateTokens(next.directory.endpoint(change), next.tokenMap.tokens(change));
105-
// needed if we miss the REGISTERED above; Does nothing if we are already in epStateMap:
106-
Gossiper.instance.maybeInitializeLocalState(SystemKeyspace.incrementAndGetGeneration());
107-
StreamSupport.stream(ColumnFamilyStore.all().spliterator(), false)
108-
.filter(cfs -> Schema.instance.getUserKeyspaces().names().contains(cfs.keyspace.getName()))
109-
.forEach(cfs -> cfs.indexManager.executePreJoinTasksBlocking(true));
110-
if (prev.directory.peerState(change) == MOVING)
111-
logger.info("Node {} state jump to NORMAL", next.directory.endpoint(change));
112-
break;
115+
// legacy log messages for tests
116+
logger.info("JOINING: Starting to bootstrap");
117+
logger.info("JOINING: calculation complete, ready to bootstrap");
113118
}
114-
// Maybe intitialise local epstate whatever the node state because we could be processing after a
115-
// replay and so may have not seen any previous local states, making this the first mutation of gossip
116-
// state for the local node.
117-
Gossiper.instance.maybeInitializeLocalState(SystemKeyspace.incrementAndGetGeneration());
118-
Gossiper.instance.addLocalApplicationState(SCHEMA, StorageService.instance.valueFactory.schema(next.schema.getVersion()));
119-
// if the local node's location has changed, update system.local.
120-
if (!next.directory.location(change).equals(prev.directory.location(change)))
121-
SystemKeyspace.updateLocation(next.directory.location(change));
122-
}
119+
break;
120+
case JOINED:
121+
tokensForGossip = next.tokenMap.tokens(localId);
122+
SystemKeyspace.updateTokens(next.directory.endpoint(localId), tokensForGossip);
123+
Set<String> userKeyspaces = Schema.instance.getUserKeyspaces().names();
124+
StreamSupport.stream(ColumnFamilyStore.all().spliterator(), false)
125+
.filter(cfs -> userKeyspaces.contains(cfs.keyspace.getName()))
126+
.forEach(cfs -> cfs.indexManager.executePreJoinTasksBlocking(true));
127+
NodeState previousState = prev.directory.peerState(localId);
128+
if (previousState == MOVING)
129+
{
130+
logger.info("Node {} state jump to NORMAL", next.directory.endpoint(localId));
131+
}
132+
else if (previousState == BOOT_REPLACING)
133+
{
134+
// legacy log message for compatibility (& tests)
135+
MultiStepOperation<?> sequence = prev.inProgressSequences.get(localId);
136+
if (sequence != null && sequence.kind() == MultiStepOperation.Kind.REPLACE)
137+
{
138+
logCompletedReplacement(prev.directory, (BootstrapAndReplace) sequence);
139+
tokensForGossip = GossipHelper.getTokensFromOperation(sequence);
140+
}
141+
}
142+
break;
143+
case MOVING:
144+
logger.debug("Node {} state MOVING, tokens {}", next.directory.endpoint(localId), prev.tokenMap.tokens(localId));
145+
tokensForGossip = next.tokenMap.tokens(localId);
146+
break;
147+
case LEFT:
148+
tokensForGossip = prev.tokenMap.tokens(localId);
149+
break;
150+
}
123151

152+
// Maybe initialise local epstate whatever the node state because we could be processing after a
153+
// replay and so may have not seen any previous local states, making this the first mutation of gossip
154+
// state for the local node.
155+
Gossiper.instance.maybeInitializeLocalState(SystemKeyspace.incrementAndGetGeneration());
156+
Gossiper.instance.addLocalApplicationState(SCHEMA, StorageService.instance.valueFactory.schema(next.schema.getVersion()));
157+
// Pull node properties from cluster metadata into gossip, except if the node is only in the REGISTERED state
158+
// as that has no equivalent gossip STATUS
159+
if (state != REGISTERED)
160+
Gossiper.instance.mergeNodeToGossip(localId, next, tokensForGossip);
161+
// if the local node's location has changed, update system.local.
162+
if (!next.directory.location(localId).equals(prev.directory.location(localId)))
163+
SystemKeyspace.updateLocation(next.directory.location(localId));
164+
}
165+
166+
private void processChangesToRemotePeers(ClusterMetadata prev, ClusterMetadata next, Set<NodeId> changed)
167+
{
168+
for (NodeId change : changed)
169+
{
170+
logger.info("Processing changes to peer {} for epoch {}->{}", change, prev.epoch.getEpoch(), next.epoch.getEpoch());
124171
if (next.directory.peerState(change) == REGISTERED)
125172
{
126173
// Re-establish any connections made prior to this node registering
@@ -155,21 +202,11 @@ else if (NodeState.isBootstrap(next.directory.peerState(change)))
155202
}
156203
else if (prev.directory.peerState(change) == BOOT_REPLACING)
157204
{
158-
// legacy log message for compatibility (& tests)
159205
MultiStepOperation<?> sequence = prev.inProgressSequences.get(change);
160206
if (sequence != null && sequence.kind() == MultiStepOperation.Kind.REPLACE)
161207
{
162-
BootstrapAndReplace replace = (BootstrapAndReplace) sequence;
163-
InetAddressAndPort replaced = prev.directory.endpoint(replace.startReplace.replaced());
164-
InetAddressAndPort replacement = prev.directory.endpoint(change);
165-
Collection<Token> tokens = GossipHelper.getTokensFromOperation(replace);
166-
logger.info("Node {} will complete replacement of {} for tokens {}", replacement, replaced, tokens);
167-
if (!replacement.equals(replaced))
168-
{
169-
for (Token token : tokens)
170-
logger.warn("Token {} changing ownership from {} to {}", token, replaced, replacement);
171-
}
172-
Gossiper.instance.mergeNodeToGossip(change, next, tokens);
208+
logCompletedReplacement(prev.directory, (BootstrapAndReplace) sequence);
209+
Gossiper.instance.mergeNodeToGossip(change, next, GossipHelper.getTokensFromOperation(sequence));
173210
PeersTable.updateLegacyPeerTable(change, prev, next);
174211
}
175212
}
@@ -181,6 +218,29 @@ else if (prev.directory.peerState(change) == BOOT_REPLACING)
181218
}
182219
}
183220

221+
private void processRemovedNodes(Set<InetAddressAndPort> removed)
222+
{
223+
for (InetAddressAndPort remove : removed)
224+
{
225+
GossipHelper.removeAndEvict(remove);
226+
PeersTable.removeFromSystemPeersTables(remove);
227+
}
228+
}
229+
230+
private void logCompletedReplacement(Directory directory, BootstrapAndReplace sequence)
231+
{
232+
// legacy log message for compatibility (& tests)
233+
InetAddressAndPort replaced = directory.endpoint(sequence.startReplace.replaced());
234+
InetAddressAndPort replacement = directory.endpoint(sequence.startReplace.replacement());
235+
Collection<Token> tokens = GossipHelper.getTokensFromOperation(sequence);
236+
logger.info("Node {} will complete replacement of {} for tokens {}", replacement, replaced, tokens);
237+
if (!replacement.equals(replaced))
238+
{
239+
for (Token token : tokens)
240+
logger.warn("Token {} changing ownership from {} to {}", token, replaced, replacement);
241+
}
242+
}
243+
184244
private boolean directoryEntryChangedFor(NodeId nodeId, Directory prev, Directory next)
185245
{
186246
return prev.peerState(nodeId) != next.peerState(nodeId) ||

‎src/java/org/apache/cassandra/tcm/listeners/UpgradeMigrationListener.java‎

Lines changed: 27 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -24,18 +24,39 @@
2424
import org.apache.cassandra.gms.Gossiper;
2525
import org.apache.cassandra.tcm.ClusterMetadata;
2626
import org.apache.cassandra.tcm.Epoch;
27+
import org.apache.cassandra.utils.CassandraVersion;
2728

28-
public class UpgradeMigrationListener implements ChangeListener.Async
29+
/**
30+
* For handling changes in Cassandra version.
31+
* One use case is to react to the initial migration from Gossip based metadata in Cassandra 5.0 and earlier. When
32+
* a node first transitions to using ClusterMetadataService, this listener will update its gossip state with the new
33+
* nodeId based hostId and ensure that is propagated.
34+
*
35+
* Another use is for evolving distributed system tables, this listener can identify when a new Cassandra version has
36+
* been deployed across the cluster and provides a hook to take actions such as creating new internal tables etc.
37+
*/
38+
public class UpgradeMigrationListener implements ChangeListener
2939
{
3040
private static final Logger logger = LoggerFactory.getLogger(UpgradeMigrationListener.class);
3141
public void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot)
3242
{
33-
if (!prev.epoch.equals(Epoch.UPGRADE_GOSSIP))
43+
if (prev.epoch.equals(Epoch.UPGRADE_GOSSIP))
44+
{
45+
logger.info("Detected upgrade from gossip mode, updating my host id in gossip to {}", next.myNodeId());
46+
Gossiper.instance.mergeNodeToGossip(next.myNodeId(), next);
47+
if (Gossiper.instance.getQuarantineDisabled())
48+
Gossiper.instance.clearQuarantinedEndpoints();
3449
return;
50+
}
3551

36-
logger.info("Detected upgrade from gossip mode, updating my host id in gossip to {}", next.myNodeId());
37-
Gossiper.instance.mergeNodeToGossip(next.myNodeId(), next);
38-
if (Gossiper.instance.getQuarantineDisabled())
39-
Gossiper.instance.clearQuarantinedEndpoints();
52+
CassandraVersion prevMinVersion = prev.directory.clusterMinVersion.cassandraVersion;
53+
CassandraVersion minVersion = next.directory.clusterMinVersion.cassandraVersion;
54+
if (prevMinVersion.compareTo(minVersion) == 0 || (prev.epoch.is(Epoch.EMPTY) && fromSnapshot))
55+
{
56+
// nothing to do if the min version in the cluster has not changed
57+
// likewise, we don't need to trigger if applying a snapshot to a previously empty cluster metadata for e.g.
58+
// when replaying at startup
59+
logger.debug("Cluster min version has not changed, nothing to do");
60+
}
4061
}
4162
}

0 commit comments

Comments
 (0)