diff --git a/CHANGES.txt b/CHANGES.txt index a63247a414..8a3d06fbff 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 9a998662e7..932423d248 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -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; diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 1e897d95a8..011fa51f20 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -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; diff --git a/src/java/org/apache/cassandra/db/Mutation.java b/src/java/org/apache/cassandra/db/Mutation.java index 5a41271474..384f2b4972 100644 --- a/src/java/org/apache/cassandra/db/Mutation.java +++ b/src/java/org/apache/cassandra/db/Mutation.java @@ -150,6 +150,11 @@ public class Mutation implements IMutation, Supplier return modifications.values(); } + public long getApproxCreatedAtNanos() + { + return approxCreatedAtNanos; + } + @Override public Supplier hintOnFailure() { diff --git a/src/java/org/apache/cassandra/hints/Hint.java b/src/java/org/apache/cassandra/hints/Hint.java index e2e74ee6f5..886e781ee6 100644 --- a/src/java/org/apache/cassandra/hints/Hint.java +++ b/src/java/org/apache/cassandra/hints/Hint.java @@ -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) diff --git a/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java b/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java index 4ebca9a232..2b7342d571 100644 --- a/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java +++ b/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java @@ -266,6 +266,11 @@ public abstract class AbstractWriteResponseHandler implements RequestCallback */ protected abstract int ackCount(); + public Dispatcher.RequestTime getRequestTime() + { + return requestTime; + } + /** * 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 cd03fae552..31e60d68e6 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -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) diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 3817077df0..5e16bae944 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -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() { diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index 3e556db04a..f2b75195c1 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -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. * diff --git a/test/distributed/org/apache/cassandra/distributed/test/AbstractHintWindowTest.java b/test/distributed/org/apache/cassandra/distributed/test/AbstractHintWindowTest.java index ecb17f8c12..5e441e27c1 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/AbstractHintWindowTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/AbstractHintWindowTest.java @@ -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) 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..21177f20cd --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/HintDataReappearingTest.java @@ -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 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 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 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()); + } + } +}