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

Merge branch 'cassandra-6.0' into trunk

This commit is contained in:
Sam Tunnicliffe 2026-06-17 12:33:24 +01:00
parent 017276263a 802ce7f8b2
commit 38e0277503
7 changed files with 215 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -3,6 +3,7 @@
* Allow nodetool garbagecollect to take a user defined list of SSTables (CASSANDRA-16767)
* Add a guardrail for misprepared statements (CASSANDRA-21139)
Merged from 6.0:
* Always send TCM commit failures as Messaging failures (CASSANDRA-21457)
* Fix ReadCommand serializedSize() using incorrect epoch (CASSANDRA-21438)
* Allocation improvements in ProtocolVersion, StorageProxy and MerkleTree (CASSANDRA-21199)
* Don’t leave autocompaction disabled during bootstrap and replace (CASSANDRA-21236)

6
src/java/org/apache/cassandra/net/Message.java
View File

@ -254,6 +254,12 @@ public class Message<T> implements ResponseContext
return outWithParam(nextId(), verb, 0, payload, flag.addTo(0), null, null);
}
public static <T> Message<T> outWithFlag(Verb verb, T payload, MessageFlag flag, long expireAtNanos)
{
assert !verb.isResponse();
return outWithParam(nextId(), verb, expireAtNanos, payload, flag.addTo(0), null, null);
}
public static <T> Message<T> outWithFlag(Verb verb, T payload, Dispatcher.RequestTime requestTime, MessageFlag flag)
{
return outWithFlags(verb, payload, requestTime, flag.addTo(0));

21
src/java/org/apache/cassandra/tcm/Commit.java
View File

@ -27,6 +27,8 @@ import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.utils.Invariants;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.exceptions.ExceptionCode;
import org.apache.cassandra.io.IVersionedSerializer;
@ -281,7 +283,7 @@ public class Commit
}
else
{
assert t instanceof Failure;
Invariants.require(t instanceof Failure);
Failure failure = (Failure) t;
out.writeByte(failure.rejected ? REJECTED : FAILED);
out.writeUnsignedVInt32(failure.code.value);
@ -327,9 +329,10 @@ public class Commit
}
else
{
assert t instanceof Failure;
size += VIntCoding.computeUnsignedVIntSize(((Failure) t).code.value);
size += TypeSizes.sizeof(((Failure)t).message);
Invariants.require(t instanceof Failure);
Failure failure = (Failure) t;
size += VIntCoding.computeUnsignedVIntSize(failure.code.value);
size += TypeSizes.sizeof(failure.message);
size += VIntCoding.computeUnsignedVIntSize(serializationVersion.asInt());
size += LogState.metadataSerializer.serializedSize(t.logState(), serializationVersion);
}
@ -375,15 +378,9 @@ public class Commit
Result.Success success = result.success();
replicator.send(success, message.from());
logger.info("Responding with full result {} to sender {}", result, message.from());
// TODO: this response message can get lost; how do we re-discover this on the other side?
// TODO: what if we have holes after replaying?
messagingService.accept(message.responseWith(result), message.from());
}
else
{
Result.Failure failure = result.failure();
messagingService.accept(message.responseWith(failure), message.from());
}
messagingService.accept(message.responseWith(result), message.from());
}
private void checkCMSState()

10
src/java/org/apache/cassandra/tcm/PeerLogFetcher.java
View File

@ -92,11 +92,11 @@ public class PeerLogFetcher
logger.info("Fetching log from {}, at least {}", remote, awaitAtleast);
try (Timer.Context ctx = TCMMetrics.instance.fetchPeerLogLatency.time())
{
RemoteProcessor.sendWithCallbackAsync(fetchRes,
Verb.TCM_FETCH_PEER_LOG_REQ,
new FetchPeerLog(before),
new RemoteProcessor.CandidateIterator(Collections.singletonList(remote), false),
Retry.untilElapsed(DatabaseDescriptor.getCmsAwaitTimeout().to(TimeUnit.NANOSECONDS), TCMMetrics.instance.fetchLogRetries));
RemoteProcessor.sendWithRetries(Verb.TCM_FETCH_PEER_LOG_REQ,
new FetchPeerLog(before),
fetchRes,
new RemoteProcessor.CandidateIterator(Collections.singletonList(remote), false),
Retry.untilElapsed(DatabaseDescriptor.getCmsAwaitTimeout().to(TimeUnit.NANOSECONDS), TCMMetrics.instance.fetchLogRetries));
return fetchRes.map((logState) -> {
log.append(logState);

161
src/java/org/apache/cassandra/tcm/RemoteProcessor.java
View File

@ -27,6 +27,7 @@ import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;
import java.util.function.Supplier;
import com.codahale.metrics.Timer;
@ -34,6 +35,7 @@ import com.codahale.metrics.Timer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.exceptions.RequestFailure;
import org.apache.cassandra.exceptions.RequestFailureReason;
@ -42,15 +44,16 @@ import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.metrics.TCMMetrics;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessageDelivery;
import org.apache.cassandra.net.MessageFlag;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.RequestCallbackWithFailure;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.service.WaitStrategy;
import org.apache.cassandra.tcm.Discovery.DiscoveredNodes;
import org.apache.cassandra.tcm.log.Entry;
import org.apache.cassandra.tcm.log.LocalLog;
import org.apache.cassandra.tcm.log.LogState;
import org.apache.cassandra.utils.AbstractIterator;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.concurrent.AsyncPromise;
import org.apache.cassandra.utils.concurrent.Future;
@ -78,10 +81,14 @@ public final class RemoteProcessor implements Processor
{
try
{
Commit.Result result = sendWithCallback(Verb.TCM_COMMIT_REQ,
new Commit(entryId, transform, lastKnown),
new CandidateIterator(candidates(false)),
retryPolicy);
Commit.Result result = sendWithRetries(Verb.TCM_COMMIT_REQ,
new Commit(entryId, transform, lastKnown),
(Commit.Result res) -> {
if (res.isFailure() && !res.failure().rejected)
throw new IllegalArgumentException(res.failure().message);
},
new CandidateIterator(candidates(false)),
retryPolicy);
log.append(result.logState());
@ -171,11 +178,11 @@ public final class RemoteProcessor implements Processor
{
Promise<LogState> remoteRequest = new AsyncPromise<>();
Epoch currentEpoch = log.metadata().epoch;
sendWithCallbackAsync(remoteRequest,
Verb.TCM_FETCH_CMS_LOG_REQ,
new FetchCMSLog(currentEpoch, ClusterMetadataService.state() == REMOTE),
candidates,
Retry.withNoTimeLimit(TCMMetrics.instance.fetchLogRetries));
sendWithRetries(Verb.TCM_FETCH_CMS_LOG_REQ,
new FetchCMSLog(currentEpoch, ClusterMetadataService.state() == REMOTE),
remoteRequest,
candidates,
Retry.withNoTimeLimit(TCMMetrics.instance.fetchLogRetries));
return remoteRequest.map((replay) -> {
if (!replay.isEmpty())
{
@ -188,14 +195,13 @@ public final class RemoteProcessor implements Processor
}
}
// todo rename to send with retries or something
public static <REQ, RSP> RSP sendWithCallback(Verb verb, REQ request, CandidateIterator candidates, WaitStrategy backoff)
public static <REQ, RSP> RSP sendWithRetries(Verb verb, REQ request, Consumer<RSP> check, CandidateIterator candidates, Retry retry)
{
Promise<RSP> future = AsyncPromise.uncancellable();
sendWithRetries(verb, request, check, future, candidates, retry);
try
{
Promise<RSP> promise = new AsyncPromise<>();
sendWithCallbackAsync(promise, verb, request, candidates, backoff);
return promise.await().get();
return future.get();
}
catch (InterruptedException | ExecutionException e)
{
@ -203,47 +209,92 @@ public final class RemoteProcessor implements Processor
}
}
public static <REQ, RSP> void sendWithCallbackAsync(Promise<RSP> promise, Verb verb, REQ request, CandidateIterator candidates, WaitStrategy backoff)
public static <REQ, RSP> void sendWithRetries(Verb verb, REQ request, Promise<RSP> future, CandidateIterator candidates, Retry retry)
{
//TODO (now): the retry defines how long to wait for a retry, but the old behavior scheduled the message right away... should this be delayed as well?
MessagingService.instance().<REQ, RSP>sendWithRetries(backoff, MessageDelivery.ImmediateRetryScheduler.instance,
verb, request, candidates,
(attempt, success, failure) -> {
if (failure != null) promise.tryFailure(failure);
else promise.trySuccess(success.payload);
},
(attempt, from, failure) -> {
if (promise.isDone() || promise.isCancelled())
return false;
if (failure.reason == RequestFailureReason.NOT_CMS)
{
logger.debug("{} is not a member of the CMS, querying it to discover current membership", from);
DiscoveredNodes cms = tryDiscover(from);
candidates.addCandidates(cms);
candidates.timeout(from);
logger.debug("Got CMS from {}: {}, retrying on: {}", from, cms, candidates);
}
else
{
candidates.timeout(from);
logger.warn("Got error from {}: {} when sending {}, retrying on {}", from, failure, verb, candidates);
}
return true;
},
(attempt, reason, from, failure) -> {
switch (reason)
{
case NoMoreCandidates:
return String.format("Ran out of candidates while sending %s: %s", verb, candidates);
case GiveUp:
return String.format("Could not succeed sending %s to %s; policy %s gave up", verb, candidates, backoff);
case Interrupted:
case FailedSchedule:
return null;
default:
throw new UnsupportedOperationException(reason.name());
}
});
sendWithRetries(verb, request, ignore_ -> {}, future, candidates, retry);
}
/**
* Sends a request to given candidate nodes with retries, respecting the retry policy.
*
* If request handler expects node to be CMS, handles CMS discovery if a candidate node reports it's not a CMS member.
*/
public static <REQ, RSP> void sendWithRetries(Verb verb, REQ request, Consumer<RSP> check, Promise<RSP> future, CandidateIterator candidates, Retry retry)
{
if (!candidates.hasNext())
{
future.setFailure(new MessageDelivery.NoMoreCandidatesException(String.format("Ran out of candidates while sending %s: %s", verb, candidates)));
return;
}
else if (retry.hasExpired())
{
future.setFailure(new MessageDelivery.GivingUpException(retry.attempts(), String.format("Could not succeed sending %s to %s; policy %s gave up", verb, candidates, retry)));
return;
}
InetAddressAndPort candidate = candidates.next();
long waitNanos = Math.min(verb.expiresAfterNanos(), Math.max(0, retry.remainingNanos()));
Message<REQ> msg = Message.outWithFlag(verb, request, MessageFlag.CALL_BACK_ON_FAILURE, Clock.Global.nanoTime() + waitNanos);
MessagingService.instance().sendWithCallback(msg, candidate, new RequestCallbackWithFailure<RSP>()
{
@Override
public void onFailure(InetAddressAndPort from, RequestFailure failure)
{
switch (failure.reason)
{
case UNKNOWN:
candidates.timeout(candidate);
logger.warn("Got error from {}: {} when sending {}, retrying on {}", from, failure, verb, candidates);
break;
case TIMEOUT:
candidates.timeout(candidate);
logger.warn("Got error from {}: timeout when sending {}, retrying on {}", candidate, verb, candidates);
break;
case NOT_CMS:
logger.debug("{} is not a member of the CMS, querying it to discover current membership", from);
DiscoveredNodes cms = tryDiscover(from);
candidates.addCandidates(cms);
candidates.timeout(from);
logger.debug("Got CMS from {}: {}, retrying on: {}", from, cms, candidates);
break;
default:
// Unknown exception - add candidate back and retry
candidates.timeout(candidate);
logger.warn("Unexpected error ({}) sending {} to {}, retrying", failure, verb, candidate);
}
if (Thread.currentThread().isInterrupted()) // preserve interrupt status
{
Thread.currentThread().interrupt();
future.setFailure(new InterruptedException("Interrupted while sending " + verb));
}
else
{
long waitFor = retry.computeWait();
if (waitFor < 0)
{
// The retry strategy returns a negative wait to signal that its attempt budget is exhausted.
future.setFailure(new MessageDelivery.GivingUpException(retry.attempts(), String.format("Could not succeed sending %s to %s; policy %s exhausted attempts", verb, candidates, retry)));
return;
}
ScheduledExecutors.nonPeriodicTasks.schedule(() -> sendWithRetries(verb, request, check, future, candidates, retry), waitFor, TimeUnit.MILLISECONDS);
}
}
@Override
public void onResponse(Message<RSP> msg)
{
try
{
check.accept(msg.payload);
future.setSuccess(msg.payload);
}
catch (Throwable t)
{
onFailure(msg.from(), new RequestFailure(RequestFailureReason.UNKNOWN, t));
}
}
});
}
private static DiscoveredNodes tryDiscover(InetAddressAndPort ep)

5
src/java/org/apache/cassandra/tcm/Retry.java
View File

@ -85,6 +85,11 @@ public class Retry implements WaitStrategy
return true;
}
public long computeWait()
{
return computeWait(attempts, TimeUnit.MILLISECONDS);
}
@Override
public long computeWaitUntil(int attempts)
{

83
test/distributed/org/apache/cassandra/distributed/test/log/CommitStartupByNonCMSNodeTest.java Normal file
View File

@ -0,0 +1,83 @@
/*
* 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.distributed.test.log;
import java.util.Random;
import java.util.function.Supplier;
import org.junit.Test;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.shared.ClusterUtils;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.membership.NodeAddresses;
import org.apache.cassandra.tcm.membership.NodeVersion;
import org.apache.cassandra.tcm.transformations.Startup;
import org.apache.cassandra.utils.FBUtilities;
import static org.junit.Assert.fail;
public class CommitStartupByNonCMSNodeTest extends FuzzTestBase
{
@Test
public void commitStartupByNonCMSNode() throws Throwable
{
try (Cluster cluster = Cluster.build(4)
.withConfig(conf -> conf.set("request_timeout", "1000ms") // for TCM commit
.set("cms_retry_delay", "10ms,retries=1") // avoid retries by CMS node to avoid triggering log fetch
.set("write_request_timeout", "100ms") // time out paxos writes quickly
.with(Feature.NETWORK, Feature.GOSSIP))
.start())
{
cluster.setUncaughtExceptionsFilter(t -> t.getMessage() != null && t.getMessage().contains("There are not enough nodes in dc0 datacenter to satisfy replication factor"));
Random rnd = new Random(2);
Supplier<Integer> nodeSelector = () -> rnd.nextInt(cluster.size() - 1) + 1;
cluster.get(nodeSelector.get()).nodetoolResult("cms", "reconfigure", "3").asserts().success();
for (int i = 2; i <= 3; i++)
ClusterUtils.stopUnchecked(cluster.get(i));
Thread startNode2 = new Thread(() -> {
cluster.get(2).startup();
});
Thread commitStartupTranformation = new Thread(() -> {
cluster.get(4).runOnInstance(() -> {
try
{
ClusterMetadataService.instance().commit(new Startup(ClusterMetadata.current().myNodeId(),
new NodeAddresses(FBUtilities.getBroadcastAddressAndPort()),
NodeVersion.CURRENT));
}
catch (Throwable t)
{
fail("Should not happen");
}
});
});
commitStartupTranformation.start();
Thread.sleep(10_000);
startNode2.start();
commitStartupTranformation.join();
startNode2.join();
}
}
}