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

Merge a5dde117a9 into 8fda60880b

This commit is contained in:
Francisco Guerrero 2026-07-22 19:01:18 -07:00 committed by GitHub
parent 8fda60880b a5dde117a9
commit a23be31e9f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 279 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/java/org/apache/cassandra/config/MutationTrackingSpec.java
View File

@ -30,4 +30,10 @@ public class MutationTrackingSpec
* The interval in which the backgroun reconciliation process runs
*/
public volatile DurationSpec.LongMillisecondsBound background_reconciliation_interval = new DurationSpec.LongMillisecondsBound("1s");
/**
* Minimum time to wait before re-issuing a pull request for the same coordinator log
* during background reconciliation. Decoupled from {@link #background_reconciliation_interval}
* so that suppression spans multiple scheduler ticks.
*/
public volatile DurationSpec.LongMillisecondsBound background_reconciliation_request_cooldown = new DurationSpec.LongMillisecondsBound("3s");
}

6
src/java/org/apache/cassandra/metrics/MutationTrackingMetrics.java
View File

@ -45,6 +45,9 @@ public class MutationTrackingMetrics
public final Counter broadcastOffsetsDiscovered; // Newly-witnessed offsets discovered via broadcast
public final Counter writeTimeOffsetsDiscovered; // Newly-witnessed offsets discovered at write time
public final Counter backgroundPullRequestsSent; // PullMutationsRequest messages issued by background reconciliation
public final Counter backgroundPullRequestsSuppressed; // PullMutationsRequest sends skipped by the per-coordinator-log cooldown
public final Counter backgroundPullRequestsFailed; // PullMutationsRequest sends that failed before transmission (overload, serialization, closed connection)
public final Histogram readSummarySize; // Read summary sizes
public final Gauge<Long> unreconciledMutationCount; // Number of unreconciled mutations
public final Gauge<Long> journalDiskSpaceUsed; // Size of MutationJournal on disk
@ -55,6 +58,9 @@ public class MutationTrackingMetrics
{
broadcastOffsetsDiscovered = Metrics.counter(factory.createMetricName("BroadcastOffsetsDiscovered"));
writeTimeOffsetsDiscovered = Metrics.counter(factory.createMetricName("WriteTimeOffsetsDiscovered"));
backgroundPullRequestsSent = Metrics.counter(factory.createMetricName("BackgroundPullRequestsSent"));
backgroundPullRequestsSuppressed = Metrics.counter(factory.createMetricName("BackgroundPullRequestsSuppressed"));
backgroundPullRequestsFailed = Metrics.counter(factory.createMetricName("BackgroundPullRequestsFailed"));
readSummarySize = Metrics.histogram(factory.createMetricName("ReadSummarySize"), true);
unreconciledMutationCount = Metrics.register(
factory.createMetricName("UnreconciledMutationCount"),

126
src/java/org/apache/cassandra/replication/MutationTrackingService.java
View File

@ -65,13 +65,17 @@ import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Splitter;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.RequestFailure;
import org.apache.cassandra.exceptions.RequestFailureReason;
import org.apache.cassandra.gms.FailureDetector;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.locator.EndpointsForRange;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.metrics.MutationTrackingMetrics;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessageFlag;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.NoPayload;
import org.apache.cassandra.net.RequestCallback;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.repair.SyncTask;
import org.apache.cassandra.repair.SyncTasks;
@ -89,6 +93,7 @@ import org.apache.cassandra.tcm.listeners.ChangeListener;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.ownership.ReplicaGroups;
import org.apache.cassandra.tcm.ownership.VersionedEndpoints;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MBeanWrapper;
@ -255,6 +260,8 @@ public class MutationTrackingService implements MutationTrackingServiceMBean
if (!keyspaceShards.isEmpty() && !config.background_reconciliation_enabled)
logBackgroundReconciliationDisabledWarning(keyspaceShards.keySet());
warnIfCooldownBelowInterval();
offsetsBroadcaster.start();
offsetsPersister.start();
backgroundReconciler.start();
@ -290,6 +297,7 @@ public class MutationTrackingService implements MutationTrackingServiceMBean
logger.info("Setting mutation tracking background reconciliation interval from {} to {}",
config.background_reconciliation_interval, backgroundReconciliationInterval);
config.background_reconciliation_interval = backgroundReconciliationInterval;
warnIfCooldownBelowInterval();
}
}
@ -299,6 +307,26 @@ public class MutationTrackingService implements MutationTrackingServiceMBean
return config.background_reconciliation_interval.toMilliseconds();
}
@Override
public void setMutationTrackingBackgroundReconciliationRequestCooldownMilliseconds(long cooldownMilliseconds)
{
if (cooldownMilliseconds != config.background_reconciliation_request_cooldown.toMilliseconds())
{
DurationSpec.LongMillisecondsBound backgroundReconciliationRequestCooldown =
new DurationSpec.LongMillisecondsBound(cooldownMilliseconds, TimeUnit.MILLISECONDS);
logger.info("Setting mutation tracking background reconciliation request cooldown from {} to {}",
config.background_reconciliation_request_cooldown, backgroundReconciliationRequestCooldown);
config.background_reconciliation_request_cooldown = backgroundReconciliationRequestCooldown;
warnIfCooldownBelowInterval();
}
}
@Override
public long getMutationTrackingBackgroundReconciliationRequestCooldownMilliseconds()
{
return config.background_reconciliation_request_cooldown.toMilliseconds();
}
public void pauseOffsetBroadcast(boolean pause)
{
offsetsBroadcaster.pauseOffsetBroadcast(pause);
@ -1465,8 +1493,21 @@ public class MutationTrackingService implements MutationTrackingServiceMBean
return rows.one().getInt("host_log_id");
}
static void warnIfCooldownBelowInterval()
{
long cooldownMs = config.background_reconciliation_request_cooldown.toMilliseconds();
long intervalMs = config.background_reconciliation_interval.toMilliseconds();
if (cooldownMs < intervalMs)
logger.warn("Mutation tracking background reconciliation request cooldown ({} ms) is less than the " +
"scheduling interval ({} ms); per-coordinator-log request suppression will not span " +
"consecutive scheduler ticks.",
cooldownMs, intervalMs);
}
private static class BackgroundReconciler
{
private final Map<CoordinatorLogId, Long> lastRequestedAt = new ConcurrentHashMap<>();
void start()
{
scheduleNext();
@ -1492,16 +1533,28 @@ public class MutationTrackingService implements MutationTrackingServiceMBean
void run()
{
MutationTrackingService.instance().forEachKeyspace(this::run);
}
long now = Clock.Global.nanoTime();
long cooldownNanos = TimeUnit.MILLISECONDS.toNanos(config.background_reconciliation_request_cooldown.toMilliseconds());
private void run(KeyspaceShards shards)
{
if (config.background_reconciliation_enabled)
shards.forEachShard(this::run);
{
MutationTrackingService.instance()
.forEachKeyspace(shards -> shards.forEachShard(shard -> run(shard, now, cooldownNanos)));
}
else if (!lastRequestedAt.isEmpty())
{
// When the background reconciliation is disabled, we clean up any pending
// requests being tracked for the cool down period
lastRequestedAt.clear();
}
if (!lastRequestedAt.isEmpty())
{
lastRequestedAt.values().removeIf(timestamp -> now - timestamp > cooldownNanos);
}
}
private void run(Shard shard)
private void run(Shard shard, long now, long cooldownNanos)
{
try
{
@ -1510,8 +1563,16 @@ public class MutationTrackingService implements MutationTrackingServiceMBean
for (Offsets.Immutable offsets : missing)
{
// Prefer pulling from the coordinator
int coordinatorHostId = offsets.logId().hostId();
CoordinatorLogId logId = offsets.logId();
Long lastRequested = lastRequestedAt.get(logId);
if (lastRequested != null && now - lastRequested < cooldownNanos)
{
MutationTrackingMetrics.instance().backgroundPullRequestsSuppressed.inc();
continue;
}
int coordinatorHostId = logId.hostId();
InetAddressAndPort coordinator = ClusterMetadata.current().directory.endpoint(new NodeId(coordinatorHostId));
InetAddressAndPort pullFrom = FailureDetector.instance.isAlive(coordinator)
? coordinator
@ -1523,10 +1584,12 @@ public class MutationTrackingService implements MutationTrackingServiceMBean
continue; // No reachable source
}
// TODO (expected): backoff, rate limits, per host and total
PullMutationsRequest request = new PullMutationsRequest(offsets, ActiveLogReconciler.Priority.REGULAR);
logger.trace("Requesting pull mutation request from replica {} for missing offset {}", pullFrom, offsets);
MessagingService.instance().send(Message.out(Verb.MT_PULL_MUTATIONS_REQ, request), pullFrom);
Message<PullMutationsRequest> message = Message.outWithFlag(Verb.MT_PULL_MUTATIONS_REQ, request, MessageFlag.CALL_BACK_ON_FAILURE);
MessagingService.instance().sendWithCallback(message, pullFrom, new PullRequestFailureCallback(logId, now));
lastRequestedAt.put(logId, now);
MutationTrackingMetrics.instance().backgroundPullRequestsSent.inc();
}
}
catch (Throwable throwable)
@ -1550,6 +1613,49 @@ public class MutationTrackingService implements MutationTrackingServiceMBean
}
return null;
}
/**
* MT_PULL_MUTATIONS_REQ is a one-way verb (no response). The callback registered with
* {@link MessagingService#sendWithCallback} fires on send-layer failures (queue overload,
* serialization error, closed connection) and on the eventual TIMEOUT after
* {@code readTimeout} expires. We treat TIMEOUT as benign (a one-way verb never produces
* a response) and only react to genuine send failures by clearing the cooldown entry,
* so the next reconciliation tick can retry without waiting for the cooldown to expire.
*/
private final class PullRequestFailureCallback implements RequestCallback<NoPayload>
{
private final CoordinatorLogId logId;
private final long sentAt;
PullRequestFailureCallback(CoordinatorLogId logId, long sentAt)
{
this.logId = logId;
this.sentAt = sentAt;
}
@Override
public void onResponse(Message<NoPayload> msg)
{
// MT_PULL_MUTATIONS_REQ is one-way; no response is expected.
}
@Override
public boolean invokeOnFailure()
{
return true;
}
@Override
public void onFailure(InetAddressAndPort from, RequestFailure failure)
{
if (failure.reason == RequestFailureReason.TIMEOUT)
return; // expected for a one-way verb
// Only remove if our entry is still the one in place; a newer reconcile may have
// already overwritten it.
lastRequestedAt.remove(logId, sentAt);
MutationTrackingMetrics.instance().backgroundPullRequestsFailed.inc();
}
}
}
// TODO (later): a more intelligent heuristic for offsets included in broadcasts

13
src/java/org/apache/cassandra/replication/MutationTrackingServiceMBean.java
View File

@ -46,4 +46,17 @@ public interface MutationTrackingServiceMBean
* @return the interval, in milliseconds, in which the background reconciliation runs when enabled
*/
long getMutationTrackingBackgroundReconciliationIntervalMilliseconds();
/**
* Sets the minimum delay (in milliseconds) between successive pull requests issued by
* background reconciliation for the same coordinator log.
*
* @param cooldownMilliseconds the cooldown value in milliseconds
*/
void setMutationTrackingBackgroundReconciliationRequestCooldownMilliseconds(long cooldownMilliseconds);
/**
* @return the per-coordinator-log request cooldown, in milliseconds, used by background reconciliation
*/
long getMutationTrackingBackgroundReconciliationRequestCooldownMilliseconds();
}

10
src/java/org/apache/cassandra/tools/NodeProbe.java
View File

@ -2879,6 +2879,16 @@ public class NodeProbe implements AutoCloseable
{
mutationTrackingProxy.setMutationTrackingBackgroundReconciliationIntervalMilliseconds(intervalMilliseconds);
}
public long getMutationTrackingBackgroundReconciliationRequestCooldownMilliseconds()
{
return mutationTrackingProxy.getMutationTrackingBackgroundReconciliationRequestCooldownMilliseconds();
}
public void setMutationTrackingBackgroundReconciliationRequestCooldownMilliseconds(long cooldownMilliseconds)
{
mutationTrackingProxy.setMutationTrackingBackgroundReconciliationRequestCooldownMilliseconds(cooldownMilliseconds);
}
}
class ColumnFamilyStoreMBeanIterator implements Iterator<Map.Entry<String, ColumnFamilyStoreMBean>>

6
src/java/org/apache/cassandra/tools/nodetool/MTAdmin.java
View File

@ -62,6 +62,7 @@ public class MTAdmin extends AbstractCommand
out.println("background_reconciliation_enabled: " + probe.getMutationTrackingBackgroundReconciliationEnabled());
out.println("background_reconciliation_interval_ms: " + probe.getMutationTrackingBackgroundReconciliationIntervalMilliseconds());
out.println("background_reconciliation_request_cooldown_ms: " + probe.getMutationTrackingBackgroundReconciliationRequestCooldownMilliseconds());
}
}
@ -73,7 +74,7 @@ public class MTAdmin extends AbstractCommand
@Parameters(index = "0", arity = "0..1", description = { "Mutation tracking param type.",
"Possible parameters: " +
"[background_reconciliation_enabled|background_reconciliation_interval_ms]" })
"[background_reconciliation_enabled|background_reconciliation_interval_ms|background_reconciliation_request_cooldown_ms]" })
public String paramType;
@Parameters(index = "1", description = "Mutation tracking param value", arity = "0..1")
@ -104,6 +105,9 @@ public class MTAdmin extends AbstractCommand
case "background_reconciliation_interval_ms":
probe.setMutationTrackingBackgroundReconciliationIntervalMilliseconds(Long.parseLong(value));
break;
case "background_reconciliation_request_cooldown_ms":
probe.setMutationTrackingBackgroundReconciliationRequestCooldownMilliseconds(Long.parseLong(value));
break;
default:
throw new IllegalArgumentException("Unknown parameter: " + type);
}

123
test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingTest.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.distributed.test.tracking;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.function.LongSupplier;
import org.junit.Assert;
import org.junit.Ignore;
@ -38,6 +39,7 @@ import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.hints.HintsService;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.metrics.MutationTrackingMetrics;
import org.apache.cassandra.metrics.StorageMetrics;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.replication.CoordinatorLogId;
@ -857,4 +859,125 @@ public class MutationTrackingTest extends TestBaseImpl
assertEquals(0, numLogReconciliations(cluster.get(3)));
}
}
@Test
public void testBackgroundReconciliationCooldown() throws Throwable
{
try (Cluster cluster = Cluster.build(3)
.withConfig(cfg -> cfg.with(Feature.NETWORK)
.with(Feature.GOSSIP)
.set("write_request_timeout", "1000ms"))
.start())
{
cluster.schemaChange(withKeyspace("CREATE KEYSPACE %s WITH replication = " +
"{'class': 'SimpleStrategy', 'replication_factor': 3} " +
"AND replication_type='tracked'"));
cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (k int PRIMARY KEY, v int)"));
// 1. Create a missing offset on node 3 using the same recipe as
// testBackgroundPullReconciliation: partition node 3, write at QUORUM, then
// reconnect and broadcast offsets so node 3 learns about the gap.
cluster.filters().allVerbs().to(3).drop();
cluster.filters().allVerbs().from(3).drop();
for (int i = 1; i <= 2; i++)
cluster.get(i).runOnInstance(() -> Gossiper.instance.convict(InetAddressAndPort.getByNameUnchecked("127.0.0.3"), Double.MAX_VALUE));
for (int i = 1; i <= 3; i++)
cluster.get(i).runOnInstance(() -> {
MutationTrackingService.instance().pauseActiveReconciler();
MutationTrackingService.instance().pauseBackgroundReconciler();
});
awaitNodeDead(cluster.get(1), cluster.get(3));
cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.tbl (k, v) VALUES (1, 1)"),
ConsistencyLevel.QUORUM);
TimeUnit.SECONDS.sleep(1);
cluster.filters().reset();
awaitNodeAlive(cluster.get(1), cluster.get(3));
awaitNodeAlive(cluster.get(3), cluster.get(1));
for (int i = 1; i <= 3; i++)
cluster.get(i).runOnInstance(() -> MutationTrackingService.instance().broadcastOffsetsForTesting());
// 2. Use a short request cooldown on node 3 so the test is fast, and keep it
// distinct from the schedule interval so the cooldown is actually exercised
// (not aliased onto the scheduling cadence).
//
// NOTE: we deliberately leave the active reconciler PAUSED on node 1 (and 2)
// so the pull requests we send from node 3 are never served — the missing
// offsets stay missing across phases, and each reconcileForTesting() that
// isn't suppressed by the cooldown produces a fresh outbound pull request.
long cooldownMs = 500;
cluster.get(3).runOnInstance(() ->
MutationTrackingService.instance().setMutationTrackingBackgroundReconciliationRequestCooldownMilliseconds(cooldownMs));
// 3. Use the BackgroundPullRequestsSent metric on node 3 as the source of truth
// for outbound pull request counts — it's incremented in lock-step with each
// actual send and avoids needing a separate counting message filter.
IInvokableInstance node3 = cluster.get(3);
LongSupplier sentCount = () ->
node3.callOnInstance(() -> MutationTrackingMetrics.instance().backgroundPullRequestsSent.getCount());
// === Phase 1: rapid-fire dedup ===
// Two reconcileForTesting calls back-to-back within the cooldown window must
// produce only ONE outbound pull request.
node3.runOnInstance(() -> {
MutationTrackingService.instance().resumeBackgroundReconciler();
MutationTrackingService.instance().reconcileForTesting();
MutationTrackingService.instance().reconcileForTesting();
MutationTrackingService.instance().pauseBackgroundReconciler();
});
TimeUnit.MILLISECONDS.sleep(200);
assertEquals("Rapid-fire dedup: only the first reconcile should send a request",
1L, sentCount.getAsLong());
// Verify the suppression code path was actually taken on the second call.
long suppressedAfterPhase1 = node3.callOnInstance(() ->
MutationTrackingMetrics.instance().backgroundPullRequestsSuppressed.getCount());
assertTrue("Cooldown suppression metric should advance when a duplicate reconcile is suppressed",
suppressedAfterPhase1 >= 1);
// === Phase 2: cooldown expires, allowing a fresh request ===
// Sleep longer than the configured cooldown. While paused, the scheduled task
// fires and exercises the disable-clear branch (and the time-based removeIf),
// so the previous entry is gone by the time we manually reconcile again.
TimeUnit.MILLISECONDS.sleep(cooldownMs + 200);
node3.runOnInstance(() -> {
MutationTrackingService.instance().resumeBackgroundReconciler();
MutationTrackingService.instance().reconcileForTesting();
MutationTrackingService.instance().pauseBackgroundReconciler();
});
assertEquals("Post-cooldown: a fresh reconcile should send a request once the cooldown elapses",
2L, sentCount.getAsLong());
// === Phase 3: disabling reconciliation clears tracked state ===
// Phase 2 left a fresh entry in lastRequestedAt. Without sleeping past the
// cooldown, invoking run() while disabled takes the disable-clear branch and
// wipes the map. The next reconcile (after re-enabling) should then send a
// request even though we are still inside Phase 2's cooldown window.
node3.runOnInstance(() -> {
// Already paused at the end of Phase 2 — explicitly run() to exercise the
// disable-clear branch with the entry still inside the cooldown window.
MutationTrackingService.instance().reconcileForTesting();
MutationTrackingService.instance().resumeBackgroundReconciler();
MutationTrackingService.instance().reconcileForTesting();
MutationTrackingService.instance().pauseBackgroundReconciler();
});
TimeUnit.MILLISECONDS.sleep(200);
// Without the disable-clear branch, the Phase 2 entry would still be within
// its cooldown and would suppress this reconcile.
assertEquals("Disabling should clear tracked-request state and allow a fresh send within the cooldown window",
3L, sentCount.getAsLong());
// Sanity: the happy path should never count a send-layer failure. The callback
// we install on each pull request only counts non-TIMEOUT failures (queue overload,
// serialization, closed connection); the eventual TIMEOUT for an unanswered one-way
// request is benign and must not advance this counter.
long failedAtEnd = node3.callOnInstance(() ->
MutationTrackingMetrics.instance().backgroundPullRequestsFailed.getCount());
assertEquals("Happy-path test should not record any send-layer failures",
0, failedAtEnd);
}
}
}