From 4f0ad22f122f8233aa99545bc234083690f53ea7 Mon Sep 17 00:00:00 2001 From: Matt Byrd Date: Thu, 7 Nov 2024 09:37:24 -0800 Subject: [PATCH] Use mutation creation time for hint expiry rather than hint submission time patch by Matt Byrd; reviewed by Chris Lohfink, reviewed by Blake Eggleston for CASSANDRA-20014 --- CHANGES.txt | 1 + .../org/apache/cassandra/config/Config.java | 1 + .../cassandra/config/DatabaseDescriptor.java | 10 + .../org/apache/cassandra/db/Mutation.java | 5 + src/java/org/apache/cassandra/hints/Hint.java | 4 +- .../service/AbstractWriteResponseHandler.java | 5 + .../cassandra/service/StorageProxy.java | 14 +- .../cassandra/service/StorageService.java | 28 +- .../service/StorageServiceMBean.java | 10 + .../test/AbstractHintWindowTest.java | 100 +++++++ .../test/HintDataReappearingTest.java | 257 ++++++++++++++++++ 11 files changed, 430 insertions(+), 5 deletions(-) create mode 100644 test/distributed/org/apache/cassandra/distributed/test/AbstractHintWindowTest.java create mode 100644 test/distributed/org/apache/cassandra/distributed/test/HintDataReappearingTest.java diff --git a/CHANGES.txt b/CHANGES.txt index ef327e6be9..67ad77673e 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0.16 + * Make hint expiry use request start time rather than timeout time for TTL (CASSANDRA-20014) * Do not attach rows and partitions to CheckForAbort when already attached (CASSANDRA-20135) * Allow hint delivery during schema mismatch (CASSANDRA-20188) * Fix gossip issue with gossip-only and bootstrapping nodes missing DC/Rack/Host ID endpoint state (CASSANDRA-19983) diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 57388eed68..24df356a4b 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -300,6 +300,7 @@ public class Config public int max_hints_file_size_in_mb = 128; public ParameterizedClass hints_compression; 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; diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index bf96a3f885..593235d577 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -2638,6 +2638,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; diff --git a/src/java/org/apache/cassandra/db/Mutation.java b/src/java/org/apache/cassandra/db/Mutation.java index 8a1ffc123c..78291e07be 100644 --- a/src/java/org/apache/cassandra/db/Mutation.java +++ b/src/java/org/apache/cassandra/db/Mutation.java @@ -147,6 +147,11 @@ public class Mutation implements IMutation return table == null ? null : modifications.get(table.id); } + public long getApproxCreatedAtNanos() + { + return approxCreatedAtNanos; + } + public boolean isEmpty() { return modifications.isEmpty(); diff --git a/src/java/org/apache/cassandra/hints/Hint.java b/src/java/org/apache/cassandra/hints/Hint.java index 6c7c5d46ba..06541dfdf4 100644 --- a/src/java/org/apache/cassandra/hints/Hint.java +++ b/src/java/org/apache/cassandra/hints/Hint.java @@ -51,13 +51,13 @@ import static org.apache.cassandra.db.TypeSizes.sizeofUnsignedVInt; * 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 = Integer.getInteger("cassandra.maxHintTTL", Integer.MAX_VALUE); + public static volatile int maxHintTTL = Integer.getInteger("cassandra.maxHintTTL", Integer.MAX_VALUE); final Mutation mutation; final long creationTime; // time of hint creation (in milliseconds) diff --git a/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java b/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java index 0ef944c247..f0d7766fa7 100644 --- a/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java +++ b/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java @@ -237,6 +237,11 @@ public abstract class AbstractWriteResponseHandler implements RequestCallback */ protected abstract int ackCount(); + public long getQueryStartNanoTime() + { + return queryStartNanoTime; + } + /** * null message means "response from local write" */ diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 3e4292ae7b..23ee648454 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -130,6 +130,9 @@ import org.apache.cassandra.utils.UUIDGen; import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.NANOSECONDS; + +import static org.apache.cassandra.utils.MonotonicClock.approxTime; + import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.casReadMetrics; import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.casWriteMetrics; import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.readMetrics; @@ -2427,8 +2430,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, System.currentTimeMillis())); + + long creationTime = System.currentTimeMillis(); + if (DatabaseDescriptor.isUseCreationTimeForHintTtl()) + { + long mutationCreationTimeNanos = responseHandler != null ? responseHandler.getQueryStartNanoTime() : 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) diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 5b29e4bee9..d3dab125e4 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -139,6 +139,7 @@ import org.apache.cassandra.gms.IEndpointStateChangeSubscriber; import org.apache.cassandra.gms.IFailureDetector; import org.apache.cassandra.gms.TokenSerializer; import org.apache.cassandra.gms.VersionedValue; +import org.apache.cassandra.hints.Hint; import org.apache.cassandra.hints.HintsService; import org.apache.cassandra.io.sstable.SSTableLoader; import org.apache.cassandra.io.sstable.format.SSTableFormat; @@ -6036,12 +6037,37 @@ public class StorageService extends NotificationBroadcasterSupport implements IE logger.info("updated transfer_hints_on_decommission to {}", enabled); } - public void setHintedHandoffThrottleInKB(int throttleInKB) + @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); + } + + public void setHintedHandoffThrottleInKB(int throttleInKB) { DatabaseDescriptor.setHintedHandoffThrottleInKB(throttleInKB); logger.info("updated hinted_handoff_throttle_in_kb to {}", throttleInKB); } + @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() { diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index 0cc6dbd21a..ecdfb0e67c 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -777,6 +777,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. * diff --git a/test/distributed/org/apache/cassandra/distributed/test/AbstractHintWindowTest.java b/test/distributed/org/apache/cassandra/distributed/test/AbstractHintWindowTest.java new file mode 100644 index 0000000000..fbb2e63116 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/AbstractHintWindowTest.java @@ -0,0 +1,100 @@ +/* + * 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.UUID; + +import org.junit.Ignore; + +import org.apache.cassandra.auth.CassandraRoleManager; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.apache.cassandra.distributed.api.IIsolatedExecutor; +import org.apache.cassandra.gms.FailureDetector; +import org.apache.cassandra.hints.HintsService; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.metrics.StorageMetrics; +import org.apache.cassandra.service.StorageService; + +import static java.util.concurrent.TimeUnit.MINUTES; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.apache.cassandra.distributed.api.ConsistencyLevel.ONE; +import static org.awaitility.Awaitility.await; + +@Ignore +public abstract class AbstractHintWindowTest extends TestBaseImpl +{ + + void pauseHintsDelivery(IInvokableInstance node) + { + node.runOnInstance((IIsolatedExecutor.SerializableRunnable) () -> { + HintsService.instance.pauseDispatch(); + }); + } + + 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) + .timeout(1, MINUTES) + .until(() -> node.appliesOnInstance((IIsolatedExecutor.SerializableBiFunction) (secondNode, online) -> { + InetAddressAndPort address = StorageService.instance.getEndpointForHostId(secondNode); + return online == FailureDetector.instance.isAlive(address); + }).apply(node2UUID, shouldBeOnline)); + } + + Long getTotalHintsCount(IInvokableInstance node) + { + return node.callOnInstance(() -> StorageMetrics.totalHints.getCount()); + } + + Long insertData(final Cluster cluster) { + + // insert data and sleep every 10k to have a chance to flush hints + for (int i = 0; i < 70000; i++) { + cluster.coordinator(1) + .execute(withKeyspace("INSERT INTO %s.cf (k, c1) VALUES (?, ?);"), + ONE, UUID.randomUUID().toString(), UUID.randomUUID().toString()); + + if (i % 10000 == 0) + await().atLeast(2, SECONDS).pollDelay(2, SECONDS).until(() -> true); + } + + await().atLeast(3, SECONDS).pollDelay(3, SECONDS).until(() -> true); + + // we see that metrics are updated + + await().until(() -> cluster.get(1).callOnInstance(() -> StorageMetrics.totalHints.getCount()) > 0); + return cluster.get(1).callOnInstance(() -> StorageMetrics.totalHints.getCount()); + } + + static void waitForExistingRoles(Cluster cluster) + { + cluster.forEach(instance -> await().pollDelay(1, SECONDS) + .pollInterval(1, SECONDS) + .atMost(60, SECONDS) + .until(() -> instance.callOnInstance(CassandraRoleManager::hasExistingRoles))); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/HintDataReappearingTest.java b/test/distributed/org/apache/cassandra/distributed/test/HintDataReappearingTest.java new file mode 100644 index 0000000000..e4df267eb8 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/HintDataReappearingTest.java @@ -0,0 +1,257 @@ +/* + * 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.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; + +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_in_ms", "30000")) + .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 keys = IntStream.range(0, 1).mapToObj(x -> UUID.randomUUID()).collect(Collectors.toList()); + + scheduler.submit(() -> { + if (dropTwoWrites) + { + Assertions.assertThatThrownBy(() -> insertData(cluster, keys)) + .isInstanceOf(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 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)); + + // 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 selectData(Cluster cluster, List keys) + { + List 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 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 insertedKeys) + { + + for (UUID partitionKey : insertedKeys) + { + cluster.coordinator(1) + .execute(withKeyspace("DELETE c1 FROM %s.cf where k=?;"), + LOCAL_QUORUM, partitionKey.toString()); + } + } +} +