diff --git a/CHANGES.txt b/CHANGES.txt index 87fb75b057..5a53faf257 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 5.0-beta2 + * Resolve the oldest hints just from descriptors and current writer if available (CASSANDRA-19600) * Optionally fail writes when SAI refuses to index a term value exceeding configured term max size (CASSANDRA-19493) * Vector search can restrict on clustering keys when filtering isn't required (CASSANDRA-19544) * Fix FBUtilities' parsing of gcp cos_containerd kernel versions (CASSANDRA-18594) diff --git a/src/java/org/apache/cassandra/hints/HintsBuffer.java b/src/java/org/apache/cassandra/hints/HintsBuffer.java index 07daa30c76..646dd72feb 100644 --- a/src/java/org/apache/cassandra/hints/HintsBuffer.java +++ b/src/java/org/apache/cassandra/hints/HintsBuffer.java @@ -19,20 +19,22 @@ package org.apache.cassandra.hints; import java.io.IOException; import java.nio.ByteBuffer; -import java.util.*; +import java.util.Collections; +import java.util.Iterator; +import java.util.Queue; +import java.util.Set; +import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicLong; import java.util.zip.CRC32; -import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.io.util.DataOutputBuffer; import org.apache.cassandra.io.util.DataOutputBufferFixed; import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.utils.AbstractIterator; -import org.apache.cassandra.utils.Clock; import org.apache.cassandra.utils.concurrent.OpOrder; import static org.apache.cassandra.utils.FBUtilities.updateChecksum; @@ -60,7 +62,6 @@ final class HintsBuffer private final ConcurrentMap> offsets; private final OpOrder appendOrder; - private final ConcurrentMap earliestHintByHost; // Stores time of the earliest hint in the buffer for each host private HintsBuffer(ByteBuffer slab) { @@ -69,7 +70,6 @@ final class HintsBuffer position = new AtomicLong(); offsets = new ConcurrentHashMap<>(); appendOrder = new OpOrder(); - earliestHintByHost = new ConcurrentHashMap<>(); } static HintsBuffer create(int slabSize) @@ -127,7 +127,7 @@ final class HintsBuffer if (bufferOffsets == null) return Collections.emptyIterator(); - return new AbstractIterator() + return new AbstractIterator<>() { private final ByteBuffer flyweight = slab.duplicate(); @@ -140,26 +140,11 @@ final class HintsBuffer int totalSize = slab.getInt(offset) + ENTRY_OVERHEAD_SIZE; - return (ByteBuffer) flyweight.clear().position(offset).limit(offset + totalSize); + return flyweight.clear().position(offset).limit(offset + totalSize); } }; } - /** - * Retrieve the time of the earliest hint in the buffer for a specific node - * @param hostId UUID of the node - * @return timestamp for the earliest hint in the buffer, or {@link Global#currentTimeMillis()} - */ - long getEarliestHintTime(UUID hostId) - { - return earliestHintByHost.getOrDefault(hostId, Clock.Global.currentTimeMillis()); - } - - void clearEarliestHintForHostId(UUID hostId) - { - earliestHintByHost.remove(hostId); - } - Allocation allocate(int hintSize) { int totalSize = hintSize + ENTRY_OVERHEAD_SIZE; @@ -240,15 +225,8 @@ final class HintsBuffer void write(Iterable hostIds, Hint hint) { write(hint); - long ts = Clock.Global.currentTimeMillis(); for (UUID hostId : hostIds) - { - // We only need the time of the first hint in the buffer - if (DatabaseDescriptor.hintWindowPersistentEnabled()) - earliestHintByHost.putIfAbsent(hostId, ts); - put(hostId, offset); - } } public void close() @@ -258,7 +236,7 @@ final class HintsBuffer private void write(Hint hint) { - ByteBuffer buffer = (ByteBuffer) slab.duplicate().position(offset).limit(offset + totalSize); + ByteBuffer buffer = slab.duplicate().position(offset).limit(offset + totalSize); CRC32 crc = new CRC32(); int hintSize = totalSize - ENTRY_OVERHEAD_SIZE; try (DataOutputBuffer dop = new DataOutputBufferFixed(buffer)) diff --git a/src/java/org/apache/cassandra/hints/HintsBufferPool.java b/src/java/org/apache/cassandra/hints/HintsBufferPool.java index b2bebde456..275dbc37e6 100644 --- a/src/java/org/apache/cassandra/hints/HintsBufferPool.java +++ b/src/java/org/apache/cassandra/hints/HintsBufferPool.java @@ -18,7 +18,6 @@ package org.apache.cassandra.hints; import java.io.Closeable; -import java.util.Iterator; import java.util.UUID; import java.util.concurrent.BlockingQueue; @@ -66,31 +65,6 @@ final class HintsBufferPool implements Closeable } } - /** - * Get the earliest hint for a specific node from all buffers - * @param hostId UUID of the node - * @return timestamp for the earliest hint - */ - long getEarliestHintForHost(UUID hostId) - { - long min = currentBuffer().getEarliestHintTime(hostId); - Iterator it = reserveBuffers.iterator(); - - while (it.hasNext()) - min = Math.min(min, it.next().getEarliestHintTime(hostId)); - - return min; - } - - public void clearEarliestHintsForHostId(UUID hostId) - { - currentBuffer().clearEarliestHintForHostId(hostId); - Iterator it = reserveBuffers.iterator(); - - while (it.hasNext()) - it.next().clearEarliestHintForHostId(hostId); - } - private HintsBuffer.Allocation allocate(int hintSize) { HintsBuffer current = currentBuffer(); diff --git a/src/java/org/apache/cassandra/hints/HintsDispatchTrigger.java b/src/java/org/apache/cassandra/hints/HintsDispatchTrigger.java index a5216429b1..0dfc6e1323 100644 --- a/src/java/org/apache/cassandra/hints/HintsDispatchTrigger.java +++ b/src/java/org/apache/cassandra/hints/HintsDispatchTrigger.java @@ -73,8 +73,6 @@ final class HintsDispatchTrigger implements Runnable if (store.isWriting()) writeExecutor.closeWriter(store); - - HintsService.instance.getHintsBufferPool().clearEarliestHintsForHostId(store.hostId); } private boolean isScheduled(HintsStore store) diff --git a/src/java/org/apache/cassandra/hints/HintsService.java b/src/java/org/apache/cassandra/hints/HintsService.java index 9c86f0bc97..19989a6c87 100644 --- a/src/java/org/apache/cassandra/hints/HintsService.java +++ b/src/java/org/apache/cassandra/hints/HintsService.java @@ -36,7 +36,6 @@ import com.google.common.collect.ImmutableMap; import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.io.util.File; import org.apache.cassandra.locator.ReplicaLayout; -import org.apache.cassandra.utils.Clock; import org.apache.cassandra.utils.concurrent.Future; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -399,8 +398,6 @@ public final class HintsService implements HintsServiceMBean // delete all the hints files and remove the HintsStore instance from the map in the catalog catalog.exciseStore(hostId); - - bufferPool.clearEarliestHintsForHostId(hostId); } /** @@ -447,17 +444,15 @@ public final class HintsService implements HintsServiceMBean } /** - * Get the earliest hint written for a particular node, - * @param hostId UUID of the node to check it's hints. - * @return earliest hint as per unix time or Long.MIN_VALUE if hostID is null + * Find the oldest hint written for a particular node by looking into descriptors + * and current open writer, if any. + * + * @param hostId UUID of the node to check its hints. + * @return the oldest hint of the given host id or Long.MAX_VALUE when not found */ - public long getEarliestHintForHost(UUID hostId) + public long findOldestHintTimestamp(UUID hostId) { - // Need to check only the first descriptor + all buffers. - HintsStore store = catalog.get(hostId); - HintsDescriptor desc = store.getFirstDescriptor(); - long timestamp = desc == null ? Clock.Global.currentTimeMillis() : desc.timestamp; - return Math.min(timestamp, bufferPool.getEarliestHintForHost(hostId)); + return catalog.get(hostId).findOldestHintTimestamp(); } HintsCatalog getCatalog() @@ -478,9 +473,4 @@ public final class HintsService implements HintsServiceMBean { return isDispatchPaused.get(); } - - HintsBufferPool getHintsBufferPool() - { - return bufferPool; - } } diff --git a/src/java/org/apache/cassandra/hints/HintsStore.java b/src/java/org/apache/cassandra/hints/HintsStore.java index 348bc8ed2c..969f37ae7f 100644 --- a/src/java/org/apache/cassandra/hints/HintsStore.java +++ b/src/java/org/apache/cassandra/hints/HintsStore.java @@ -124,6 +124,25 @@ final class HintsStore return new PendingHintsInfo(hostId, queueSize, minTimestamp, maxTimestamp); } + /** + * Find the oldest hint written for a particular node by looking into descriptors + * and current open writer, if any. + * + * @return the oldest hint as per unix time or Long.MAX_VALUE if not present + */ + public long findOldestHintTimestamp() + { + HintsDescriptor desc = dispatchDequeue.peekFirst(); + if (desc != null) + return desc.timestamp; + + HintsWriter writer = getWriter(); + if (writer != null) + return writer.descriptor().timestamp; + + return Long.MAX_VALUE; + } + boolean isLive() { InetAddressAndPort address = address(); @@ -262,14 +281,6 @@ final class HintsStore corruptedFiles.add(descriptor); } - /** - * @return a copy of the first {@link HintsDescriptor} in the queue for dispatch or {@code null} if queue is empty. - */ - HintsDescriptor getFirstDescriptor() - { - return dispatchDequeue.peekFirst(); - } - /* * Methods dealing with HintsWriter. * diff --git a/src/java/org/apache/cassandra/hints/HintsWriteExecutor.java b/src/java/org/apache/cassandra/hints/HintsWriteExecutor.java index 8456091074..2b0a434dd3 100644 --- a/src/java/org/apache/cassandra/hints/HintsWriteExecutor.java +++ b/src/java/org/apache/cassandra/hints/HintsWriteExecutor.java @@ -221,36 +221,34 @@ final class HintsWriteExecutor private void flush(Iterator iterator, HintsStore store) { - while (true) + while (iterator.hasNext()) { - if (iterator.hasNext()) - flushInternal(iterator, store); - - if (!iterator.hasNext()) - break; - - // exceeded the size limit for an individual file, but still have more to write - // close the current writer and continue flushing to a new one in the next iteration - store.closeWriter(); + // If we exceed the size limit for a hints file then close the current writer, + // if we still have more to write, we'll open a new file in the next iteration. + if (!flushInternal(iterator, store.getOrOpenWriter())) + store.closeWriter(); } } - private void flushInternal(Iterator iterator, HintsStore store) + /** + * @return {@code true} if we can keep writing to the file, + * or {@code false} if we've exceeded max file size limit during writing + */ + private boolean flushInternal(Iterator iterator, HintsWriter writer) { long maxHintsFileSize = DatabaseDescriptor.getMaxHintsFileSize(); - HintsWriter writer = store.getOrOpenWriter(); - try (HintsWriter.Session session = writer.newSession(writeBuffer)) { while (iterator.hasNext()) { - // check that we are not over the limit already - if (session.position() >= maxHintsFileSize) - break; - session.append(iterator.next()); + + if (session.position() >= maxHintsFileSize) + return false; } + + return true; } catch (IOException e) { diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index bfd30d8738..2f3d8683f7 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -2437,10 +2437,10 @@ public class StorageProxy implements StorageProxyMBean // if persisting hints window, hintWindowExpired might be updated according to the timestamp of the earliest hint if (tryEnablePersistentWindow && !hintWindowExpired && DatabaseDescriptor.hintWindowPersistentEnabled()) { - long earliestHint = HintsService.instance.getEarliestHintForHost(hostIdForEndpoint); - hintWindowExpired = Clock.Global.currentTimeMillis() - maxHintWindow > earliestHint; + long oldestHint = HintsService.instance.findOldestHintTimestamp(hostIdForEndpoint); + hintWindowExpired = Clock.Global.currentTimeMillis() - maxHintWindow > oldestHint; if (hintWindowExpired) - Tracing.trace("Not hinting {} for which there is the earliest hint stored at {}", replica, earliestHint); + Tracing.trace("Not hinting {} for which there is the oldest hint stored at {}", replica, oldestHint); } if (hintWindowExpired) 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..ecb17f8c12 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/AbstractHintWindowTest.java @@ -0,0 +1,126 @@ +/* + * 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.io.util.File; +import org.apache.cassandra.io.util.FileUtils; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.metrics.StorageMetrics; +import org.apache.cassandra.service.StorageService; +import org.assertj.core.api.Assertions; + +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.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +@Ignore +public abstract class AbstractHintWindowTest extends TestBaseImpl +{ + void waitUntilNoHints(IInvokableInstance node1, UUID node2UUID) + { + await().pollInterval(10, SECONDS) + .timeout(1, MINUTES) + .until(() -> getTotalHintsSize(node1, node2UUID) == 0); + } + + Long getTotalHintsSize(IInvokableInstance node, UUID node2UUID) + { + return node.appliesOnInstance((IIsolatedExecutor.SerializableFunction) secondNode -> { + return HintsService.instance.getTotalHintsSize(secondNode); + }).apply(node2UUID); + } + + void pauseHintsDelivery(IInvokableInstance node) + { + node.runOnInstance((IIsolatedExecutor.SerializableRunnable) () -> { + HintsService.instance.pauseDispatch(); + }); + } + + 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()); + } + + void assertHintsSizes(IInvokableInstance node, UUID node2UUID) + { + // we indeed have some hints in its dir + File hintsDir = new File(node.config().getString("hints_directory")); + assertThat(FileUtils.folderSize(hintsDir)).isPositive(); + + // and there is positive size of hints on the disk for the second node + long totalHintsSize = node.appliesOnInstance((IIsolatedExecutor.SerializableFunction) secondNode -> { + return HintsService.instance.getTotalHintsSize(secondNode); + }).apply(node2UUID); + + Assertions.assertThat(totalHintsSize).isPositive(); + } + + 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/HintsMaxWindowTest.java b/test/distributed/org/apache/cassandra/distributed/test/HintsMaxWindowTest.java index 0b7676f184..c8ce68db51 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/HintsMaxWindowTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/HintsMaxWindowTest.java @@ -22,31 +22,19 @@ import java.util.UUID; import org.junit.Test; -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.io.util.File; -import org.apache.cassandra.io.util.FileUtils; -import org.apache.cassandra.locator.InetAddressAndPort; -import org.apache.cassandra.metrics.StorageMetrics; import org.apache.cassandra.service.StorageService; -import org.assertj.core.api.Assertions; import static java.lang.String.format; -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.apache.cassandra.distributed.api.Feature.GOSSIP; import static org.apache.cassandra.distributed.api.Feature.NETWORK; -import static org.assertj.core.api.Assertions.assertThat; -import static org.awaitility.Awaitility.await; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; -public class HintsMaxWindowTest extends TestBaseImpl +@SuppressWarnings("Convert2MethodRef") +public class HintsMaxWindowTest extends AbstractHintWindowTest { @Test public void testHintsKeepRecordingAfterNodeGoesOfflineRepeatedly() throws Exception @@ -113,76 +101,4 @@ public class HintsMaxWindowTest extends TestBaseImpl } } - private void waitUntilNoHints(IInvokableInstance node1, UUID node2UUID) - { - await().pollInterval(10, SECONDS) - .timeout(1, MINUTES) - .until(() -> getTotalHintsSize(node1, node2UUID) == 0); - } - - private Long getTotalHintsSize(IInvokableInstance node, UUID node2UUID) - { - return node.appliesOnInstance((IIsolatedExecutor.SerializableFunction) secondNode -> { - return HintsService.instance.getTotalHintsSize(secondNode); - }).apply(node2UUID); - } - - private 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)); - } - - - private Long getTotalHintsCount(IInvokableInstance node) - { - return node.callOnInstance(() -> StorageMetrics.totalHints.getCount()); - } - - private void assertHintsSizes(IInvokableInstance node, UUID node2UUID) - { - // we indeed have some hints in its dir - File hintsDir = new File(node.config().getString("hints_directory")); - assertThat(FileUtils.folderSize(hintsDir)).isPositive(); - - // and there is positive size of hints on the disk for the second node - long totalHintsSize = node.appliesOnInstance((IIsolatedExecutor.SerializableFunction) secondNode -> { - return HintsService.instance.getTotalHintsSize(secondNode); - }).apply(node2UUID); - - Assertions.assertThat(totalHintsSize).isPositive(); - } - - private 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()); - } - - private 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/HintsPersistentWindowTest.java b/test/distributed/org/apache/cassandra/distributed/test/HintsPersistentWindowTest.java new file mode 100644 index 0000000000..bd8e8b474a --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/HintsPersistentWindowTest.java @@ -0,0 +1,99 @@ +/* + * 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.Test; + +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.apache.cassandra.distributed.api.IIsolatedExecutor; +import org.apache.cassandra.service.StorageService; + +import static java.lang.String.format; +import static org.apache.cassandra.distributed.api.Feature.GOSSIP; +import static org.apache.cassandra.distributed.api.Feature.NETWORK; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotEquals; + +@SuppressWarnings("Convert2MethodRef") +public class HintsPersistentWindowTest extends AbstractHintWindowTest +{ + @Test + public void testPersistentHintWindow() throws Exception + { + try (Cluster cluster = init(Cluster.build(2) + .withDataDirCount(1) + .withConfig(config -> config.with(NETWORK, GOSSIP) + .set("hinted_handoff_enabled", true) + .set("max_hints_delivery_threads", "1") + .set("hints_flush_period", "1s") + .set("max_hint_window", "30s") + .set("max_hints_file_size", "10MiB")) + .start(), 2)) + { + final IInvokableInstance node1 = cluster.get(1); + final IInvokableInstance node2 = cluster.get(2); + + waitForExistingRoles(cluster); + + String createTableStatement = format("CREATE TABLE %s.cf (k text PRIMARY KEY, c1 text) " + + "WITH compaction = {'class': 'SizeTieredCompactionStrategy', 'enabled': 'false'} ", KEYSPACE); + cluster.schemaChange(createTableStatement); + + UUID node2UUID = node2.callOnInstance((IIsolatedExecutor.SerializableCallable) () -> StorageService.instance.getLocalHostUUID()); + + // shutdown the second node in a blocking manner + node2.shutdown().get(); + waitUntilNodeState(node1, node2UUID, false); + + Long totalHintsAfterFirstShutdown = insertData(cluster); + Long totalHitsSizeAfterFirstShutdown = getTotalHintsSize(node1, node2UUID); + + // check hints are there etc + assertHintsSizes(node1, node2UUID); + + pauseHintsDelivery(node1); + + // wait to pass max_hint_window + Thread.sleep(60000); + + // start the second node, this will not deliver hints to it from the first because dispatch is paused + // we need this in order to keep hints still on disk, so we can check that the oldest hint + // is older than max_hint_window which will not deliver any hints even the node is not down long enough + node2.startup(); + waitUntilNodeState(node1, node2UUID, true); + + Long totalHitsSizeAfterSecondShutdown = getTotalHintsSize(node1, node2UUID); + assertEquals(totalHitsSizeAfterFirstShutdown, totalHitsSizeAfterSecondShutdown); + + // stop the node again + // boolean hintWindowExpired = endpointDowntime > maxHintWindow will be false + // then persistent window kicks in, because even it has not expired, + // there are hints to be delivered on the disk which were stil not dispatched + node2.shutdown().get(); + + Long totalHintsAfterSecondShutdown = insertData(cluster); + + assertNotEquals(0L, (long) getTotalHintsSize(node1, node2UUID)); + assertEquals(totalHintsAfterFirstShutdown, totalHintsAfterSecondShutdown); + } + } +} diff --git a/test/unit/org/apache/cassandra/hints/HintsBufferTest.java b/test/unit/org/apache/cassandra/hints/HintsBufferTest.java index 8ca0b03967..9509262553 100644 --- a/test/unit/org/apache/cassandra/hints/HintsBufferTest.java +++ b/test/unit/org/apache/cassandra/hints/HintsBufferTest.java @@ -26,7 +26,6 @@ import java.util.zip.CRC32; import com.google.common.collect.Iterables; import org.junit.BeforeClass; import org.junit.Test; -import org.junit.runner.RunWith; import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.concurrent.NamedThreadFactory; @@ -41,9 +40,6 @@ import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.schema.KeyspaceParams; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.TableMetadata; -import org.apache.cassandra.utils.Clock; -import org.jboss.byteman.contrib.bmunit.BMRule; -import org.jboss.byteman.contrib.bmunit.BMUnitRunner; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; @@ -54,7 +50,6 @@ import static org.junit.Assert.fail; import static org.apache.cassandra.utils.ByteBufferUtil.bytes; import static org.apache.cassandra.utils.FBUtilities.updateChecksum; -@RunWith(BMUnitRunner.class) public class HintsBufferTest { private static final String KEYSPACE = "hints_buffer_test"; @@ -165,48 +160,6 @@ public class HintsBufferTest buffer.free(); } - static volatile long timestampForHint = 0; - // BM rule to get the timestamp that was used to store the hint so that we avoid any flakiness in timestamps between - // when we send the hint and when it actually got written. - @Test - @BMRule(name = "GetHintTS", - targetClass="HintsBuffer$Allocation", - targetMethod="write(Iterable, Hint)", - targetLocation="AFTER INVOKE putIfAbsent", - action="org.apache.cassandra.hints.HintsBufferTest.timestampForHint = $ts") - public void testEarliestHintTime() - { - int hintSize = (int) Hint.serializer.serializedSize(createHint(0, Clock.Global.currentTimeMillis()), MessagingService.current_version); - int entrySize = hintSize + HintsBuffer.ENTRY_OVERHEAD_SIZE; - // allocate a slab to fit 10 hints - int slabSize = entrySize * 10; - - // use a fixed timestamp base for all mutation timestamps - long baseTimestamp = Clock.Global.currentTimeMillis(); - - HintsBuffer buffer = HintsBuffer.create(slabSize); - UUID uuid = UUID.randomUUID(); - // Track the first hints time - try (HintsBuffer.Allocation allocation = buffer.allocate(hintSize)) - { - Hint hint = createHint(100, baseTimestamp); - allocation.write(Collections.singleton(uuid), hint); - } - long oldestHintTime = timestampForHint; - - // Write some more hints to ensure we actually test getting the earliest - for (int i = 0; i < 9; i++) - { - try (HintsBuffer.Allocation allocation = buffer.allocate(hintSize)) - { - Hint hint = createHint(i, baseTimestamp); - allocation.write(Collections.singleton(uuid), hint); - } - } - long earliest = buffer.getEarliestHintTime(uuid); - assertEquals(oldestHintTime, earliest); - } - private static int validateEntry(UUID hostId, ByteBuffer buffer, long baseTimestamp, UUID[] load) throws IOException { CRC32 crc = new CRC32(); diff --git a/test/unit/org/apache/cassandra/hints/HintsServiceTest.java b/test/unit/org/apache/cassandra/hints/HintsServiceTest.java index 10a70402f7..dd0eb5a6ed 100644 --- a/test/unit/org/apache/cassandra/hints/HintsServiceTest.java +++ b/test/unit/org/apache/cassandra/hints/HintsServiceTest.java @@ -17,8 +17,6 @@ */ package org.apache.cassandra.hints; -import java.util.Collections; -import java.util.UUID; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -31,13 +29,9 @@ import org.junit.After; import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; -import org.junit.runner.RunWith; import com.datastax.driver.core.utils.MoreFutures; import org.apache.cassandra.SchemaLoader; -import org.apache.cassandra.Util; -import org.apache.cassandra.db.DecoratedKey; -import org.apache.cassandra.db.partitions.PartitionUpdate; import org.apache.cassandra.metrics.StorageMetrics; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.net.MockMessagingService; @@ -46,16 +40,12 @@ import org.apache.cassandra.schema.KeyspaceParams; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.service.StorageService; -import org.jboss.byteman.contrib.bmunit.BMRule; -import org.jboss.byteman.contrib.bmunit.BMUnitRunner; import static org.apache.cassandra.hints.HintsTestUtil.MockFailureDetector; import static org.apache.cassandra.hints.HintsTestUtil.sendHintsAndResponses; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotEquals; import static org.junit.Assert.assertTrue; -@RunWith(BMUnitRunner.class) public class HintsServiceTest { private static final String KEYSPACE = "hints_service_test"; @@ -183,42 +173,4 @@ public class HintsServiceTest assertTrue(dispatchOffset != null); assertTrue(((ChecksummedDataInput.Position) dispatchOffset).sourcePosition > 0); } - - // BM rule to get the timestamp that was used to store the hint so that we avoid any flakiness in timestamps between - // when we send the hint and when it actually got written. - static volatile long timestampForHint = 0L; - @Test - @BMRule(name = "GetHintTS", - targetClass="HintsBuffer$Allocation", - targetMethod="write(Iterable, Hint)", - targetLocation="AFTER INVOKE putIfAbsent", - action="org.apache.cassandra.hints.HintsServiceTest.timestampForHint = $ts") - public void testEarliestHint() throws InterruptedException - { - // create and write noOfHints using service - UUID hostId = StorageService.instance.getLocalHostUUID(); - TableMetadata metadata = Schema.instance.getTableMetadata(KEYSPACE, TABLE); - - long ts = System.currentTimeMillis(); - DecoratedKey dkey = Util.dk(String.valueOf(1)); - PartitionUpdate.SimpleBuilder builder = PartitionUpdate.simpleBuilder(metadata, dkey).timestamp(ts); - builder.row("column0").add("val", "value0"); - Hint hint = Hint.create(builder.buildAsMutation(), ts); - HintsService.instance.write(hostId, hint); - long oldestHintTime = timestampForHint; - Thread.sleep(1); - HintsService.instance.write(hostId, hint); - Thread.sleep(1); - HintsService.instance.write(hostId, hint); - - // Close and fsync so that we get the timestamp from the descriptor rather than the buffer. - HintsStore store = HintsService.instance.getCatalog().get(hostId); - HintsService.instance.flushAndFsyncBlockingly(Collections.singletonList(hostId)); - store.closeWriter(); - - long earliest = HintsService.instance.getEarliestHintForHost(hostId); - assertEquals(oldestHintTime, earliest); - assertNotEquals(oldestHintTime, timestampForHint); - } - }