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

Merge branch 'cassandra-5.0' into trunk 

* cassandra-5.0:
  Use mutation creation time for hint expiry rather than hint submission time
This commit is contained in:
Blake Eggleston 2025-01-21 12:15:45 -08:00
parent 8a16f2fce8 0cc7a03523
commit 15d6437283
11 changed files with 341 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -150,6 +150,7 @@ Merged from 4.1:
* Enforce CQL message size limit on multiframe messages (CASSANDRA-20052)
* Fix race condition in DecayingEstimatedHistogramReservoir during rescale (CASSANDRA-19365)
Merged from 4.0:
* Make hint expiry use request start time rather than timeout time for TTL (CASSANDRA-20014)
* Do not attach rows and partitions to QueryCancellationChecker when already attached (CASSANDRA-20135)
* Allow hint delivery during schema mismatch (CASSANDRA-20188)
* IndexOutOfBoundsException when accessing partition where the column was deleted (CASSANDRA-20108)

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

@ -460,6 +460,7 @@ public class Config
public ParameterizedClass hints_compression;
public volatile boolean auto_hints_cleanup_enabled = false;
public volatile boolean transfer_hints_on_decommission = true;
public volatile boolean use_creation_time_for_hint_ttl = true;
public volatile boolean incremental_backups = false;
public boolean trickle_fsync = false;

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

@ -3925,6 +3925,16 @@ public class DatabaseDescriptor
conf.transfer_hints_on_decommission = enabled;
}
public static boolean isUseCreationTimeForHintTtl()
{
return conf.use_creation_time_for_hint_ttl;
}
public static void setUseCreationTimeForHintTtl(boolean enabled)
{
conf.use_creation_time_for_hint_ttl = enabled;
}
public static boolean isIncrementalBackupsEnabled()
{
return conf.incremental_backups;

5
src/java/org/apache/cassandra/db/Mutation.java
View File

@ -150,6 +150,11 @@ public class Mutation implements IMutation, Supplier<Mutation>
return modifications.values();
}
public long getApproxCreatedAtNanos()
{
return approxCreatedAtNanos;
}
@Override
public Supplier<Mutation> hintOnFailure()
{

4
src/java/org/apache/cassandra/hints/Hint.java
View File

@ -54,13 +54,13 @@ import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
* compacted away while the hint was in storage.
*
* We also look at the smallest current value of the gcgs param for each affected table when applying the hint, and use
* creation time + min(recorded gc gs, current gcgs + current gc grace) as the overall hint expiration time.
* mutation creation time + min(recorded gc gs, current gcgs + current gc grace) as the overall hint expiration time.
* This allows now to safely reduce gc gs on tables without worrying that an applied old hint might resurrect any data.
*/
public final class Hint
{
public static final Serializer serializer = new Serializer();
static final int maxHintTTL = CASSANDRA_MAX_HINT_TTL.getInt();
public static volatile int maxHintTTL = CASSANDRA_MAX_HINT_TTL.getInt();
final Mutation mutation;
final long creationTime; // time of hint creation (in milliseconds)

5
src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java
View File

@ -266,6 +266,11 @@ public abstract class AbstractWriteResponseHandler<T> implements RequestCallback
*/
protected abstract int ackCount();
public Dispatcher.RequestTime getRequestTime()
{
return requestTime;
}
/**
* null message means "response from local write"
*/

12
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -178,6 +178,7 @@ import static org.apache.cassandra.service.paxos.v1.ProposeVerbHandler.doPropose
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.apache.cassandra.utils.LocalizeString.toUpperCaseLocalized;
import static org.apache.cassandra.utils.MonotonicClock.Global.approxTime;
import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID;
import static org.apache.cassandra.utils.concurrent.CountDownLatch.newCountDownLatch;
import static org.apache.commons.lang3.StringUtils.join;
@ -2754,8 +2755,15 @@ public class StorageProxy implements StorageProxyMBean
else
logger.debug("Discarding hint for endpoint not part of ring: {}", target);
}
logger.trace("Adding hints for {}", validTargets);
HintsService.instance.write(hostIds, Hint.create(mutation, currentTimeMillis()));
long creationTime = currentTimeMillis();
if (DatabaseDescriptor.isUseCreationTimeForHintTtl())
{
long mutationCreationTimeNanos = responseHandler != null ? responseHandler.getRequestTime().startedAtNanos() : mutation.getApproxCreatedAtNanos();
creationTime -= TimeUnit.MILLISECONDS.convert(Math.max(0, approxTime.now() - mutationCreationTimeNanos), NANOSECONDS);
}
logger.trace("Adding hints for {} with creation time {} ms", validTargets, creationTime);
HintsService.instance.write(hostIds, Hint.create(mutation, creationTime));
validTargets.forEach(HintsService.instance.metrics::incrCreatedHints);
// Notify the handler only for CL == ANY
if (responseHandler != null && responseHandler.replicaPlan.consistencyLevel() == ConsistencyLevel.ANY)

27
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -124,6 +124,7 @@ import org.apache.cassandra.gms.FailureDetector;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.gms.IEndpointStateChangeSubscriber;
import org.apache.cassandra.gms.VersionedValue;
import org.apache.cassandra.hints.Hint;
import org.apache.cassandra.hints.HintsService;
import org.apache.cassandra.index.IndexStatusManager;
import org.apache.cassandra.io.sstable.IScrubber;
@ -4761,6 +4762,32 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
logger.info("updated transfer_hints_on_decommission to {}", enabled);
}
@Override
public boolean isHintTtlUseMutationCreationTime()
{
return DatabaseDescriptor.isUseCreationTimeForHintTtl();
}
@Override
public void setUseCreationTimeForHintTtl(boolean enabled)
{
DatabaseDescriptor.setUseCreationTimeForHintTtl(enabled);
logger.info("updated use_creation_time_for_hint_ttl to {}", enabled);
}
@Override
public int getMaxHintTTL()
{
return Hint.maxHintTTL;
}
@Override
public void setMaxHintTTL(int maxHintTTL)
{
Hint.maxHintTTL = maxHintTTL;
logger.info("updated Hint.maxHintTTL to {}", maxHintTTL);
}
@Override
public void clearConnectionHistory()
{

10
src/java/org/apache/cassandra/service/StorageServiceMBean.java
View File

@ -1106,6 +1106,16 @@ public interface StorageServiceMBean extends NotificationEmitter
public boolean getTransferHintsOnDecommission();
public void setTransferHintsOnDecommission(boolean enabled);
/** Returns whether we are using the creation time of the mutation for determining hint ttl **/
public boolean isHintTtlUseMutationCreationTime();
/** Sets whether we are using the creation time of the mutation for determining hint ttl **/
public void setUseCreationTimeForHintTtl(boolean enabled);
/** Returns upper bound on the hint ttl **/
public int getMaxHintTTL();
/** Sets the upper bound on the hint ttl **/
public void setMaxHintTTL(int maxHintTTL);
/**
* Resume bootstrap streaming when there is failed data streaming.
*

7
test/distributed/org/apache/cassandra/distributed/test/AbstractHintWindowTest.java
View File

@ -65,6 +65,13 @@ public abstract class AbstractHintWindowTest extends TestBaseImpl
});
}
void transferHints(IInvokableInstance node, UUID transferToNode)
{
node.runOnInstance((IIsolatedExecutor.SerializableRunnable) () -> {
HintsService.instance.transferHints(() -> transferToNode);
});
}
void waitUntilNodeState(IInvokableInstance node, UUID node2UUID, boolean shouldBeOnline)
{
await().pollInterval(10, SECONDS)

263
test/distributed/org/apache/cassandra/distributed/test/HintDataReappearingTest.java Normal file
View File

@ -0,0 +1,263 @@
/*
* 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;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.distributed.api.IIsolatedExecutor;
import org.apache.cassandra.exceptions.WriteTimeoutException;
import org.apache.cassandra.hints.HintsService;
import org.apache.commons.lang3.ArrayUtils;
import org.junit.Assert;
import org.junit.Test;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.api.SimpleQueryResult;
import org.apache.cassandra.net.Verb;
import org.assertj.core.api.Assertions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static java.lang.String.format;
import static org.apache.cassandra.distributed.api.ConsistencyLevel.ALL;
import static org.apache.cassandra.distributed.api.ConsistencyLevel.LOCAL_QUORUM;
import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL;
import static org.apache.cassandra.distributed.api.Feature.NETWORK;
import static org.apache.cassandra.utils.AssertionUtils.isThrowable;
public class HintDataReappearingTest extends AbstractHintWindowTest
{
private static final Logger logger = LoggerFactory.getLogger(HintDataReappearingTest.class);
private static final ScheduledExecutorService scheduler = Executors
.newScheduledThreadPool(1,
new ThreadFactoryBuilder()
.setNameFormat("hint reappearance test")
.setDaemon(true)
.build());
@Test
public void testHintCausesDataReappearance() throws Exception
{
doHintReappearData(true, false);
}
@Test
public void demonstrateHintCausesDataReappearance() throws Exception
{
doHintReappearData(false, false);
}
@Test
public void testHintCausesDataReappearanceWriteTimeout() throws Exception
{
doHintReappearData(true, true);
}
@Test
public void demonstrateHintCausesDataReappearanceWriteTimeout() throws Exception
{
doHintReappearData(false, true);
}
public void doHintReappearData(final boolean preventReappearance, final boolean dropTwoWrites) throws Exception
{
try (Cluster cluster = init(Cluster.build(3)
.withDataDirCount(1)
.withConfig(config -> config.with(NETWORK, GOSSIP, NATIVE_PROTOCOL)
.set("hinted_handoff_enabled", true)
.set("max_hints_delivery_threads", "1")
.set("use_creation_time_for_hint_ttl", preventReappearance ? "true" : "false")
.set("write_request_timeout", "30000ms")
.set("hints_flush_period", "1s")
.set("max_hints_file_size", "10MiB"))
.start(), 3))
{
final IInvokableInstance node1 = cluster.get(1);
final IInvokableInstance node2 = cluster.get(2);
final IInvokableInstance node3 = cluster.get(3);
waitForExistingRoles(cluster);
// We create a table with low gc_grace_seconds to ensure we can check the interactions in a timely manner
// Also make compaction fairly proactive in an attempt avoiding to force compact repeatedly
final int gc_grace_seconds = 40;
final String createTableStatement = format("CREATE TABLE %s.cf (k text PRIMARY KEY, c1 text) " +
"WITH gc_grace_seconds = " + gc_grace_seconds +
" AND compaction = {'class': 'SizeTieredCompactionStrategy', 'min_threshold': 2, 'max_threshold': 32 } ", KEYSPACE);
cluster.schemaChange(createTableStatement);
// setup a message filter to drop mutations requests from node1 to node2 so it creates hints for those mutations
AtomicBoolean dropWritesForNode2 = new AtomicBoolean(true);
// toggling this on allows the test to timeout the original write, a slight variation, both cause the problem.
AtomicBoolean dropWritesForNode3 = new AtomicBoolean(dropTwoWrites);
cluster.filters()
.verbs(Verb.MUTATION_REQ.id)
.from(1)
.messagesMatching((from, to, message) ->
(to == 2 && dropWritesForNode2.get()
|| (to == 3 && dropWritesForNode3.get())))
.drop();
logger.info("Pausing hint delivery");
// pause hint delivery to imitate hints being behind/backed up
pauseHintsDelivery(node1);
logger.info("Inserting data");
List<UUID> keys = IntStream.range(0, 1).mapToObj(x -> UUID.randomUUID()).collect(Collectors.toList());
scheduler.submit(() -> {
if (dropTwoWrites)
{
Assertions.assertThatThrownBy(() -> insertData(cluster, keys))
.is(isThrowable(WriteTimeoutException.class));
}
else
{
insertData(cluster, keys);
}
});
Thread.sleep(1000);
dropWritesForNode2.set(false);
dropWritesForNode3.set(false);
node1.flush(KEYSPACE);
node2.flush(KEYSPACE);
node3.flush(KEYSPACE);
logger.info("Deleting data");
deleteData(cluster, keys);
long afterDelete = System.currentTimeMillis();
node1.flush(KEYSPACE);
node2.flush(KEYSPACE);
node3.flush(KEYSPACE);
logger.info("Repairing");
for (IInvokableInstance node : Arrays.asList(node1, node2, node3))
{
node.nodetoolResult(ArrayUtils.addAll(new String[]{ "repair", KEYSPACE }, "--full")).asserts().success();
}
logger.info("Done repairing");
for (SimpleQueryResult result : selectData(cluster, keys))
{
Object[][] objectArrays = result.toObjectArrays();
logger.info("Result after delete: {} {}", result, Arrays.deepToString(objectArrays));
// We expect the data to appear to be deleted initially
Assert.assertNull(objectArrays[0][1]);
}
// wait to pass gc_grace_seconds with slight buffer of 2 seconds to ensure delete persisted long enough to be gced
long msSinceDelete = Math.abs(System.currentTimeMillis() - afterDelete);
long sleepFor = Math.max(0, 1000 * gc_grace_seconds + 2000 - msSinceDelete);
logger.info("Sleeping {} ms to ensure gc_grace_seconds has ellapsed after tombstone creation", sleepFor);
Thread.sleep(sleepFor);
// ensure tombstone purged on all 3 nodes
node1.forceCompact(KEYSPACE, "cf");
node3.forceCompact(KEYSPACE, "cf");
node2.forceCompact(KEYSPACE, "cf");
IIsolatedExecutor.CallableNoExcept<UUID> node2hostId = node2.callsOnInstance(SystemKeyspace::getLocalHostId);
transferHints(node1, node2hostId.call());
// Sleep a bit more to ensure hint is delivered after tombstone is GCed
Thread.sleep(200);
logger.info("Total Hints after sleeping: {}", getTotalHintsCount(node1));
String hintInfo = node1.callsOnInstance(() -> String.valueOf(HintsService.instance.getPendingHintsInfo().size())).call();
logger.info("Number of pending hints after sleeping: {}", hintInfo);
// Check the results of reading, we expect the data to remain deleted
// and hint not to cause data to be visibile again as the hint should have expired
for (SimpleQueryResult result : selectData(cluster, keys))
{
logger.info("Result: {}", result);
Object[][] objectArrays = result.toObjectArrays();
if (preventReappearance)
{
logger.info("Preventing reappearance with mutation ttl time, hence expecting null as column value");
Assert.assertNull(objectArrays[0][1]);
}
else
{
logger.info("Demonstrating reappearance possible, hence observing non null, non empty column");
Assert.assertFalse(UUID.fromString((String) objectArrays[0][1]).toString().isEmpty());
}
}
}
}
private List<SimpleQueryResult> selectData(Cluster cluster, List<UUID> keys)
{
List<SimpleQueryResult> results = new ArrayList<>();
for (UUID partitionKey : keys)
{
SimpleQueryResult result = cluster.coordinator(1)
.executeWithResult(withKeyspace("SELECT * FROM %s.cf where k=?;"),
ALL, partitionKey.toString());
results.add(result);
}
return results;
}
private void insertData(Cluster cluster, List<UUID> inserts)
{
for (int i = 0; i < inserts.size(); i++)
{
final UUID partitionKey = inserts.get(i);
cluster.coordinator(1)
.execute(withKeyspace("INSERT INTO %s.cf (k, c1) VALUES (?, ?);"),
LOCAL_QUORUM, partitionKey.toString(), UUID.randomUUID().toString());
}
}
void deleteData(Cluster cluster, List<UUID> insertedKeys)
{
for (UUID partitionKey : insertedKeys)
{
cluster.coordinator(1)
.execute(withKeyspace("DELETE c1 FROM %s.cf where k=?;"),
LOCAL_QUORUM, partitionKey.toString());
}
}
}