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

Merge branch 'cassandra-4.1' into cassandra-5.0

This commit is contained in:
David Capwell 2025-05-22 14:14:49 -07:00
parent 85a659a432 005d3c28e5
commit ac05c94e6a
7 changed files with 106 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -16,6 +16,7 @@
* Fixed multiple single-node SAI query bugs relating to static columns (CASSANDRA-20338)
* Upgrade com.datastax.cassandra:cassandra-driver-core:3.11.5 to org.apache.cassandra:cassandra-driver-core:3.12.1 (CASSANDRA-17231)
Merged from 4.0:
* Gossip doesn't converge due to race condition when updating EndpointStates multiple fields (CASSANDRA-20659)
* Honor MAX_PARALLEL_TRANSFERS correctly (CASSANDRA-20532)
* Updating a column with a new TTL but same expiration time is non-deterministic and causes repair mismatches. (CASSANDRA-20561)
* Avoid computing prepared statement size for unprepared batches (CASSANDRA-20556)

105
src/java/org/apache/cassandra/gms/EndpointState.java
View File

@ -26,6 +26,7 @@ import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -52,8 +53,19 @@ public class EndpointState
public final static IVersionedSerializer<EndpointState> serializer = new EndpointStateSerializer();
public final static IVersionedSerializer<EndpointState> nullableSerializer = NullableSerializer.wrap(serializer);
private volatile HeartBeatState hbState;
private final AtomicReference<Map<ApplicationState, VersionedValue>> applicationState;
private static class View
{
final HeartBeatState hbState;
final Map<ApplicationState, VersionedValue> applicationState;
private View(HeartBeatState hbState, Map<ApplicationState, VersionedValue> applicationState)
{
this.hbState = hbState;
this.applicationState = applicationState;
}
}
private final AtomicReference<View> ref;
/* fields below do not get serialized */
private volatile long updateTimestamp;
@ -61,18 +73,20 @@ public class EndpointState
public EndpointState(HeartBeatState initialHbState)
{
this(initialHbState, new EnumMap<ApplicationState, VersionedValue>(ApplicationState.class));
this(initialHbState, new EnumMap<>(ApplicationState.class));
}
public EndpointState(EndpointState other)
{
this(new HeartBeatState(other.hbState), new EnumMap<>(other.applicationState.get()));
ref = new AtomicReference<>(other.ref.get());
updateTimestamp = nanoTime();
isAlive = true;
}
EndpointState(HeartBeatState initialHbState, Map<ApplicationState, VersionedValue> states)
@VisibleForTesting
public EndpointState(HeartBeatState initialHbState, Map<ApplicationState, VersionedValue> states)
{
hbState = initialHbState;
applicationState = new AtomicReference<Map<ApplicationState, VersionedValue>>(new EnumMap<>(states));
ref = new AtomicReference<>(new View(initialHbState, new EnumMap<>(states)));
updateTimestamp = nanoTime();
isAlive = true;
}
@ -80,28 +94,58 @@ public class EndpointState
@VisibleForTesting
public HeartBeatState getHeartBeatState()
{
return hbState;
return ref.get().hbState;
}
void setHeartBeatState(HeartBeatState newHbState)
public void updateHeartBeat()
{
updateTimestamp();
hbState = newHbState;
updateHeartBeat(HeartBeatState::updateHeartBeat);
}
public void forceNewerGenerationUnsafe()
{
updateHeartBeat(HeartBeatState::forceNewerGenerationUnsafe);
}
@VisibleForTesting
public void forceHighestPossibleVersionUnsafe()
{
updateHeartBeat(HeartBeatState::forceHighestPossibleVersionUnsafe);
}
void unsafeSetEmptyHeartBeatState()
{
updateHeartBeat(ignore -> HeartBeatState.empty());
}
private void updateHeartBeat(Function<HeartBeatState, HeartBeatState> fn)
{
HeartBeatState previous = null;
HeartBeatState update = null;
while (true)
{
View view = ref.get();
if (previous == null || view.hbState != previous) // if this races with updating states then can avoid bumping versions
update = fn.apply(view.hbState);
if (ref.compareAndSet(view, new View(update, view.applicationState)))
return;
previous = view.hbState;
}
}
public VersionedValue getApplicationState(ApplicationState key)
{
return applicationState.get().get(key);
return ref.get().applicationState.get(key);
}
public boolean containsApplicationState(ApplicationState key)
{
return applicationState.get().containsKey(key);
return ref.get().applicationState.containsKey(key);
}
public Set<Map.Entry<ApplicationState, VersionedValue>> states()
{
return applicationState.get().entrySet();
return ref.get().applicationState.entrySet();
}
public void addApplicationState(ApplicationState key, VersionedValue value)
@ -115,17 +159,27 @@ public class EndpointState
}
public void addApplicationStates(Set<Map.Entry<ApplicationState, VersionedValue>> values)
{
addApplicationStates(values, null);
}
public void addApplicationStates(Set<Map.Entry<ApplicationState, VersionedValue>> values, @Nullable HeartBeatState hbState)
{
while (true)
{
Map<ApplicationState, VersionedValue> orig = applicationState.get();
View view = this.ref.get();
Map<ApplicationState, VersionedValue> orig = view.applicationState;
Map<ApplicationState, VersionedValue> copy = new EnumMap<>(orig);
for (Map.Entry<ApplicationState, VersionedValue> value : values)
copy.put(value.getKey(), value.getValue());
if (applicationState.compareAndSet(orig, copy))
if (this.ref.compareAndSet(view, new View(hbState == null ? view.hbState : hbState, copy)))
{
if (hbState != null)
updateTimestamp();
return;
}
}
}
@ -133,18 +187,19 @@ public class EndpointState
{
while (hasLegacyFields())
{
Map<ApplicationState, VersionedValue> orig = applicationState.get();
View view = ref.get();
Map<ApplicationState, VersionedValue> orig = view.applicationState;
Map<ApplicationState, VersionedValue> updatedStates = filterMajorVersion3LegacyApplicationStates(orig);
// avoid updating if no state is removed
if (orig.size() == updatedStates.size()
|| applicationState.compareAndSet(orig, updatedStates))
|| ref.compareAndSet(view, new View(view.hbState, updatedStates)))
return;
}
}
private boolean hasLegacyFields()
{
Set<ApplicationState> statesPresent = applicationState.get().keySet();
Set<ApplicationState> statesPresent = ref.get().applicationState.keySet();
if (statesPresent.isEmpty())
return false;
return (statesPresent.contains(ApplicationState.STATUS) && statesPresent.contains(ApplicationState.STATUS_WITH_PORT))
@ -209,7 +264,7 @@ public class EndpointState
public boolean isStateEmpty()
{
return applicationState.get().isEmpty();
return ref.get().applicationState.isEmpty();
}
/**
@ -217,14 +272,15 @@ public class EndpointState
*/
public boolean isEmptyWithoutStatus()
{
Map<ApplicationState, VersionedValue> state = applicationState.get();
View view = ref.get();
Map<ApplicationState, VersionedValue> state = view.applicationState;
boolean hasStatus = state.containsKey(ApplicationState.STATUS_WITH_PORT) || state.containsKey(ApplicationState.STATUS);
return hbState.isEmpty() && !hasStatus
return view.hbState.isEmpty() && !hasStatus
// In the very specific case where hbState.isEmpty and STATUS is missing, this is known to be safe to "fake"
// the data, as this happens when the gossip state isn't coming from the node but instead from a peer who
// restarted and is missing the node's state.
//
// When hbState is not empty, then the node gossiped an empty STATUS; this happens during bootstrap and it's not
// When hbState is not empty, then the node gossiped an empty STATUS; this happens during bootstrap, and it's not
// possible to tell if this is ok or not (we can't really tell if the node is dead or having networking issues).
// For these cases allow an external actor to verify and inform Cassandra that it is safe - this is done by
// updating the LOOSE_DEF_OF_EMPTY_ENABLED field.
@ -279,7 +335,8 @@ public class EndpointState
public String toString()
{
return "EndpointState: HeartBeatState = " + hbState + ", AppStateMap = " + applicationState.get();
View view = ref.get();
return "EndpointState: HeartBeatState = " + view.hbState + ", AppStateMap = " + view.applicationState;
}
public boolean isSupersededBy(EndpointState that)

32
src/java/org/apache/cassandra/gms/Gossiper.java
View File

@ -341,7 +341,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean,
taskLock.lock();
/* Update the local heartbeat counter. */
endpointStateMap.get(getBroadcastAddressAndPort()).getHeartBeatState().updateHeartBeat();
endpointStateMap.get(getBroadcastAddressAndPort()).updateHeartBeat();
if (logger.isTraceEnabled())
logger.trace("My heartbeat is now {}", endpointStateMap.get(FBUtilities.getBroadcastAddressAndPort()).getHeartBeatState().getHeartBeatVersion());
final List<GossipDigest> gDigests = new ArrayList<>();
@ -638,7 +638,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean,
epState.addApplicationState(ApplicationState.STATUS_WITH_PORT, shutdown);
epState.addApplicationState(ApplicationState.STATUS, StorageService.instance.valueFactory.shutdown(true));
epState.addApplicationState(ApplicationState.RPC_READY, StorageService.instance.valueFactory.rpcReady(false));
epState.getHeartBeatState().forceHighestPossibleVersionUnsafe();
epState.forceHighestPossibleVersionUnsafe();
markDead(endpoint, epState);
FailureDetector.instance.forceConviction(endpoint);
GossiperDiagnostics.markedAsShutdown(this, endpoint);
@ -664,7 +664,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean,
VersionedValue shutdown = remoteState.getApplicationState(ApplicationState.STATUS_WITH_PORT);
if (shutdown == null)
throw new AssertionError("Remote shutdown sent but missing STATUS_WITH_PORT; " + remoteState);
remoteState.getHeartBeatState().forceHighestPossibleVersionUnsafe();
remoteState.forceHighestPossibleVersionUnsafe();
endpointStateMap.put(endpoint, remoteState);
markDead(endpoint, remoteState);
FailureDetector.instance.forceConviction(endpoint);
@ -859,7 +859,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean,
// update the other node's generation to mimic it as if it had changed it itself
logger.info("Advertising removal for {}", endpoint);
epState.updateTimestamp(); // make sure we don't evict it too soon
epState.getHeartBeatState().forceNewerGenerationUnsafe();
epState.forceNewerGenerationUnsafe();
Map<ApplicationState, VersionedValue> states = new EnumMap<>(ApplicationState.class);
states.put(ApplicationState.STATUS_WITH_PORT, StorageService.instance.valueFactory.removingNonlocal(hostId));
states.put(ApplicationState.STATUS, StorageService.instance.valueFactory.removingNonlocal(hostId));
@ -879,7 +879,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean,
{
EndpointState epState = endpointStateMap.get(endpoint);
epState.updateTimestamp(); // make sure we don't evict it too soon
epState.getHeartBeatState().forceNewerGenerationUnsafe();
epState.forceNewerGenerationUnsafe();
long expireTime = computeExpireTime();
epState.addApplicationState(ApplicationState.STATUS_WITH_PORT, StorageService.instance.valueFactory.removedNonlocal(hostId, expireTime));
epState.addApplicationState(ApplicationState.STATUS, StorageService.instance.valueFactory.removedNonlocal(hostId, expireTime));
@ -930,7 +930,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean,
else if (newState.getHeartBeatState().getHeartBeatVersion() != heartbeat)
throw new RuntimeException("Endpoint still alive: " + endpoint + " heartbeat changed while trying to assassinate it");
epState.updateTimestamp(); // make sure we don't evict it too soon
epState.getHeartBeatState().forceNewerGenerationUnsafe();
epState.forceNewerGenerationUnsafe();
}
Collection<Token> tokens = null;
@ -1755,15 +1755,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean,
// don't assert here, since if the node restarts the version will go back to zero
int oldVersion = localState.getHeartBeatState().getHeartBeatVersion();
localState.setHeartBeatState(remoteState.getHeartBeatState());
if (logger.isTraceEnabled())
logger.trace("Updating heartbeat state version to {} from {} for {} ...", localState.getHeartBeatState().getHeartBeatVersion(), oldVersion, addr);
Set<Entry<ApplicationState, VersionedValue>> remoteStates = remoteState.states();
assert remoteState.getHeartBeatState().getGeneration() == localState.getHeartBeatState().getGeneration();
Set<Entry<ApplicationState, VersionedValue>> updatedStates = remoteStates.stream().filter(entry -> {
Set<Entry<ApplicationState, VersionedValue>> updatedStates = remoteState.states().stream().filter(entry -> {
// filter out the states that are already up to date (has the same or higher version)
VersionedValue local = localState.getApplicationState(entry.getKey());
return (local == null || local.version < entry.getValue().version);
@ -1776,7 +1768,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean,
logger.trace("Updating {} state version to {} for {}", entry.getKey().toString(), entry.getValue().version, addr);
}
}
localState.addApplicationStates(updatedStates);
localState.addApplicationStates(updatedStates, remoteState.getHeartBeatState());
if (logger.isTraceEnabled())
logger.trace("Updating heartbeat state version to {} from {} for {} ...", localState.getHeartBeatState().getHeartBeatVersion(), oldVersion, addr);
// get rid of legacy fields once the cluster is not in mixed mode
if (!hasMajorVersion3Nodes)
@ -2177,7 +2171,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean,
public void forceNewerGeneration()
{
EndpointState epstate = endpointStateMap.get(getBroadcastAddressAndPort());
epstate.getHeartBeatState().forceNewerGenerationUnsafe();
epstate.forceNewerGenerationUnsafe();
}
@ -2198,7 +2192,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean,
if (epState != null)
{
logger.debug("not replacing a previous epState for {}, but reusing it: {}", ep, epState);
epState.setHeartBeatState(HeartBeatState.empty());
epState.unsafeSetEmptyHeartBeatState();
}
else
{
@ -2670,7 +2664,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean,
EndpointState epState = endpointStateMap.get(getBroadcastAddressAndPort());
if (epState != null)
{
epState.getHeartBeatState().updateHeartBeat();
epState.updateHeartBeat();
if (logger.isTraceEnabled())
logger.trace("My heartbeat is now {}", epState.getHeartBeatState().getHeartBeatVersion());
}

20
src/java/org/apache/cassandra/gms/HeartBeatState.java
View File

@ -19,8 +19,6 @@ package org.apache.cassandra.gms;
import java.io.*;
import com.google.common.annotations.VisibleForTesting;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
@ -29,15 +27,14 @@ import org.apache.cassandra.io.util.DataOutputPlus;
/**
* HeartBeat State associated with any given endpoint.
*/
public class HeartBeatState
{
public static final int EMPTY_VERSION = -1;
public static final IVersionedSerializer<HeartBeatState> serializer = new HeartBeatStateSerializer();
private volatile int generation;
private volatile int version;
private final int generation;
private final int version;
HeartBeatState(int gen)
{
@ -75,9 +72,9 @@ public class HeartBeatState
return generation;
}
void updateHeartBeat()
HeartBeatState updateHeartBeat()
{
version = VersionGenerator.getNextVersion();
return new HeartBeatState(generation, VersionGenerator.getNextVersion());
}
public int getHeartBeatVersion()
@ -85,15 +82,14 @@ public class HeartBeatState
return version;
}
void forceNewerGenerationUnsafe()
HeartBeatState forceNewerGenerationUnsafe()
{
generation += 1;
return new HeartBeatState(generation + 1, version);
}
@VisibleForTesting
public void forceHighestPossibleVersionUnsafe()
HeartBeatState forceHighestPossibleVersionUnsafe()
{
version = Integer.MAX_VALUE;
return new HeartBeatState(generation, Integer.MAX_VALUE);
}
public String toString()

2
test/distributed/org/apache/cassandra/distributed/impl/UnsafeGossipHelper.java
View File

@ -269,7 +269,7 @@ public class UnsafeGossipHelper
EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(getByAddress(address));
VersionedValue status = new VersionedValue.VersionedValueFactory(partitioner).shutdown(true);
state.addApplicationState(ApplicationState.STATUS, status);
state.getHeartBeatState().forceHighestPossibleVersionUnsafe();
state.forceHighestPossibleVersionUnsafe();
StorageService.instance.onChange(getByAddress(address), ApplicationState.STATUS, status);
});
};

2
test/unit/org/apache/cassandra/gms/GossiperTest.java
View File

@ -321,7 +321,7 @@ public class GossiperTest
proposedRemoteState = new EndpointState(proposedRemoteHeartBeat);
// Bump the heartbeat version and use the same TOKENS state
proposedRemoteHeartBeat.updateHeartBeat();
proposedRemoteState.updateHeartBeat();
proposedRemoteState.addApplicationState(ApplicationState.TOKENS, tokensValue);
// The following state change should only update heartbeat without updating the TOKENS state

2
test/unit/org/apache/cassandra/gms/SerializationsTest.java
View File

@ -130,7 +130,7 @@ public class SerializationsTest extends AbstractSerializationsTester
private static List<GossipDigest> Digests = new ArrayList<GossipDigest>();
{
HeartbeatSt.updateHeartBeat();
EndpointSt.updateHeartBeat();
EndpointSt.addApplicationState(ApplicationState.LOAD, vv0);
EndpointSt.addApplicationState(ApplicationState.STATUS_WITH_PORT, vv1);
for (int i = 0; i < 100; i++)