mirror of https://github.com/apache/cassandra
Resolve the oldest hints just from descriptors and current writer if available
patch by Stefan Miklosovic; reviewed by Aleksey Yeschenko for CASSANDRA-19600
This commit is contained in:
Stefan Miklosovic
parent
e22f67a2f7
commit
326bf4b3f5
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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<UUID, Queue<Integer>> offsets;
|
||||
private final OpOrder appendOrder;
|
||||
private final ConcurrentMap<UUID, Long> 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<ByteBuffer>()
|
||||
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<UUID> 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))
|
||||
|
|
|
|||
|
|
@ -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<HintsBuffer> 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<HintsBuffer> it = reserveBuffers.iterator();
|
||||
|
||||
while (it.hasNext())
|
||||
it.next().clearEarliestHintForHostId(hostId);
|
||||
}
|
||||
|
||||
private HintsBuffer.Allocation allocate(int hintSize)
|
||||
{
|
||||
HintsBuffer current = currentBuffer();
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
*
|
||||
|
|
|
|||
|
|
@ -221,36 +221,34 @@ final class HintsWriteExecutor
|
|||
|
||||
private void flush(Iterator<ByteBuffer> 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<ByteBuffer> 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<ByteBuffer> 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)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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<UUID, Long>) 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<UUID, Boolean, Boolean>) (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<UUID, Long>) 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)));
|
||||
}
|
||||
}
|
||||
|
|
@ -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<UUID, Long>) 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<UUID, Boolean, Boolean>) (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<UUID, Long>) 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)));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<UUID>) () -> 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue