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

CEP-15: Accord metrics 

Patch by Jacek Lewandowski, reviewed by Caleb Rackliffe, David Capwell and Henrik Ingo for CASSANDRA-18580
This commit is contained in:
Jacek Lewandowski 2023-10-10 12:09:44 +02:00 committed by David Capwell
parent cffe1cc61a
commit 27a9313970
21 changed files with 1217 additions and 123 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit b1befa3cc0a8496451bb48ec3bb1c0f56b8c7653
Subproject commit 0419858bd1f6761f08fd1369477f7c142f5bbb4f

4
src/java/org/apache/cassandra/metrics/AccordClientRequestMetrics.java
View File

@ -19,20 +19,17 @@
package org.apache.cassandra.metrics;
import com.codahale.metrics.Histogram;
import com.codahale.metrics.Meter;
import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
public class AccordClientRequestMetrics extends ClientRequestMetrics
{
public final Meter preempts;
public final Histogram keySize;
public AccordClientRequestMetrics(String scope)
{
super(scope);
preempts = Metrics.meter(factory.createMetricName("Preempts"));
keySize = Metrics.histogram(factory.createMetricName("KeySizeHistogram"), false);
}
@ -40,7 +37,6 @@ public class AccordClientRequestMetrics extends ClientRequestMetrics
public void release()
{
super.release();
Metrics.remove(factory.createMetricName("Preempts"));
Metrics.remove(factory.createMetricName("KeySizeHistogram"));
}
}

312
src/java/org/apache/cassandra/metrics/AccordMetrics.java Normal file
View File

@ -0,0 +1,312 @@
/*
* 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.metrics;
import java.lang.reflect.Field;
import java.util.concurrent.TimeUnit;
import accord.api.EventsListener;
import accord.local.Command;
import accord.primitives.Deps;
import accord.primitives.PartialDeps;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import com.codahale.metrics.Counting;
import com.codahale.metrics.Histogram;
import com.codahale.metrics.Meter;
import com.codahale.metrics.Timer;
import org.apache.cassandra.service.accord.AccordService;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
public class AccordMetrics
{
public final static AccordMetrics readMetrics = new AccordMetrics("ro");
public final static AccordMetrics writeMetrics = new AccordMetrics("rw");
public static final String COMMIT_LATENCY = "CommitLatency";
public static final String EXECUTE_LATENCY = "ExecuteLatency";
public static final String APPLY_LATENCY = "ApplyLatency";
public static final String APPLY_DURATION = "ApplyDuration";
public static final String PARTIAL_DEPENDENCIES = "PartialDependencies";
public static final String PROGRESS_LOG_SIZE = "ProgressLogSize";
public static final String DEPENDENCIES = "Dependencies";
public static final String FAST_PATHS = "FastPaths";
public static final String SLOW_PATHS = "SlowPaths";
public static final String PREEMPTS = "Preempts";
public static final String TIMEOUTS = "Timeouts";
public static final String INVALIDATIONS = "Invalidations";
public static final String RECOVERY_DELAY = "RecoveryDelay";
public static final String RECOVERY_TIME = "RecoveryTime";
public static final String FAST_PATH_TO_TOTAL = "FastPathToTotal";
public static final String ACCORD_REPLICA = "accord-replica";
public static final String ACCORD_COORDINATOR = "accord-coordinator";
/**
* The time between start on the coordinator and commit on this replica.
*/
public final Timer commitLatency;
/**
* The time between start on the coordinator and execution on this replica.
*/
public final Timer executeLatency;
/**
* The time between start on the coordinator and application on this replica.
*/
public final Timer applyLatency;
/**
* Duration of applying changes.
*/
public final Timer applyDuration;
/**
* A histogram of the number of dependencies per partial transaction at this replica.
*/
public final Histogram partialDependencies;
public final Meter progressLogSize;
/**
* A histogram of the number of dependencies per transaction at this coordinator.
*/
public final Histogram dependencies;
/**
* The number of fast path transactions executed on this coordinator.
*/
public final Meter fastPaths;
/**
* The number of slow path transactions executed on this coordinator.
*/
public final Meter slowPaths;
/**
* The number of preempted transactions on this coordinator.
*/
public final Meter preempts;
/**
* The number of timed out transactions on this coordinator.
*/
public final Meter timeouts;
/**
* The number of invalidated transactions on this coordinator.
*/
public final Meter invalidations;
/**
* The time between the start of the transaction and the start of the recovery, if the transaction is recovered.
*/
public final Timer recoveryDelay;
/**
* The time between the start of the recovery and the execution of the transaction, if the transaction is recovered.
*/
public final Timer recoveryDuration;
/**
* The ratio of the number of fast path transactions to the total number of transactions.
*/
public final RatioGaugeSet fastPathToTotal;
private AccordMetrics(String scope)
{
DefaultNameFactory replica = new DefaultNameFactory(ACCORD_REPLICA, scope);
commitLatency = Metrics.timer(replica.createMetricName(COMMIT_LATENCY));
executeLatency = Metrics.timer(replica.createMetricName(EXECUTE_LATENCY));
applyLatency = Metrics.timer(replica.createMetricName(APPLY_LATENCY));
applyDuration = Metrics.timer(replica.createMetricName(APPLY_DURATION));
partialDependencies = Metrics.histogram(replica.createMetricName(PARTIAL_DEPENDENCIES), true);
progressLogSize = Metrics.meter(replica.createMetricName(PROGRESS_LOG_SIZE));
DefaultNameFactory coordinator = new DefaultNameFactory(ACCORD_COORDINATOR, scope);
dependencies = Metrics.histogram(coordinator.createMetricName(DEPENDENCIES), true);
fastPaths = Metrics.meter(coordinator.createMetricName(FAST_PATHS));
slowPaths = Metrics.meter(coordinator.createMetricName(SLOW_PATHS));
preempts = Metrics.meter(coordinator.createMetricName(PREEMPTS));
timeouts = Metrics.meter(coordinator.createMetricName(TIMEOUTS));
invalidations = Metrics.meter(coordinator.createMetricName(INVALIDATIONS));
recoveryDelay = Metrics.timer(coordinator.createMetricName(RECOVERY_DELAY));
recoveryDuration = Metrics.timer(coordinator.createMetricName(RECOVERY_TIME));
fastPathToTotal = new RatioGaugeSet(fastPaths, RatioGaugeSet.sum(fastPaths, slowPaths), coordinator, FAST_PATH_TO_TOTAL + ".%s");
}
@Override
public String toString()
{
StringBuilder builder = new StringBuilder();
builder.append("AccordMetrics [");
try
{
for (Field f : getClass().getDeclaredFields())
{
f.setAccessible(true);
if (Counting.class.isAssignableFrom(f.getType()))
{
Counting metric = (Counting) f.get(this);
builder.append(String.format("%s: count=%d, ", f.getName(), metric.getCount()));
}
}
}
catch (IllegalAccessException e)
{
throw new RuntimeException(e);
}
builder.append("]");
return builder.toString();
}
public static class Listener implements EventsListener
{
public final static Listener instance = new Listener(AccordMetrics.readMetrics, AccordMetrics.writeMetrics);
private final AccordMetrics readMetrics;
private final AccordMetrics writeMetrics;
public Listener(AccordMetrics readMetrics, AccordMetrics writeMetrics)
{
this.readMetrics = readMetrics;
this.writeMetrics = writeMetrics;
}
private AccordMetrics forTransaction(TxnId txnId)
{
if (txnId.isWrite())
return writeMetrics;
else if (txnId.isRead())
return readMetrics;
else
return null;
}
@Override
public void onCommitted(Command cmd)
{
long now = AccordService.uniqueNow();
AccordMetrics metrics = forTransaction(cmd.txnId());
if (metrics != null)
{
long trxTimestamp = cmd.txnId().hlc();
metrics.commitLatency.update(now - trxTimestamp, TimeUnit.MICROSECONDS);
}
}
@Override
public void onExecuted(Command cmd)
{
long now = AccordService.uniqueNow();
AccordMetrics metrics = forTransaction(cmd.txnId());
if (metrics != null)
{
Timestamp trxTimestamp = cmd.txnId();
metrics.executeLatency.update(now - trxTimestamp.hlc(), TimeUnit.MICROSECONDS);
PartialDeps deps = cmd.partialDeps();
metrics.partialDependencies.update(deps != null ? deps.txnIdCount() : 0);
}
}
@Override
public void onApplied(Command cmd, long applyStartTimestamp)
{
long now = AccordService.uniqueNow();
AccordMetrics metrics = forTransaction(cmd.txnId());
if (metrics != null)
{
Timestamp trxTimestamp = cmd.txnId();
metrics.applyLatency.update(now - trxTimestamp.hlc(), TimeUnit.MICROSECONDS);
metrics.applyDuration.update(now - applyStartTimestamp, TimeUnit.MICROSECONDS);
}
}
@Override
public void onFastPathTaken(TxnId txnId, Deps deps)
{
AccordMetrics metrics = forTransaction(txnId);
if (metrics != null)
{
metrics.fastPaths.mark();
metrics.dependencies.update(deps.txnIdCount());
}
}
@Override
public void onSlowPathTaken(TxnId txnId, Deps deps)
{
AccordMetrics metrics = forTransaction(txnId);
if (metrics != null)
{
metrics.slowPaths.mark();
metrics.dependencies.update(deps.txnIdCount());
}
}
@Override
public void onRecover(TxnId txnId, Timestamp recoveryTimestamp)
{
AccordMetrics metrics = forTransaction(txnId);
if (metrics != null)
{
long now = AccordService.uniqueNow();
metrics.recoveryDuration.update(now - recoveryTimestamp.hlc(), MICROSECONDS);
metrics.recoveryDelay.update(recoveryTimestamp.hlc() - txnId.hlc(), MICROSECONDS);
}
}
@Override
public void onPreempted(TxnId txnId)
{
AccordMetrics metrics = forTransaction(txnId);
if (metrics != null)
metrics.preempts.mark();
}
@Override
public void onTimeout(TxnId txnId)
{
AccordMetrics metrics = forTransaction(txnId);
if (metrics != null)
metrics.timeouts.mark();
}
@Override
public void onInvalidated(TxnId txnId)
{
AccordMetrics metrics = forTransaction(txnId);
if (metrics != null)
metrics.invalidations.mark();
}
@Override
public void onProgressLogSizeChange(TxnId txnId, int delta)
{
AccordMetrics metrics = forTransaction(txnId);
if (metrics != null)
metrics.progressLogSize.mark(delta);
}
}
}

50
src/java/org/apache/cassandra/metrics/AccordStateCacheMetrics.java Normal file
View File

@ -0,0 +1,50 @@
/*
* 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.metrics;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import com.codahale.metrics.Histogram;
import static org.apache.cassandra.metrics.CacheMetrics.TYPE_NAME;
import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
public class AccordStateCacheMetrics extends CacheAccessMetrics
{
public static final String OBJECT_SIZE = "ObjectSize";
public final Histogram objectSize;
private final Map<Class<?>, CacheAccessMetrics> instanceMetrics = new ConcurrentHashMap<>(2);
private final String type;
public AccordStateCacheMetrics(String type)
{
super(new DefaultNameFactory(TYPE_NAME, type));
objectSize = Metrics.histogram(factory.createMetricName(OBJECT_SIZE), false);
this.type = type;
}
public CacheAccessMetrics forInstance(Class<?> klass)
{
return instanceMetrics.computeIfAbsent(klass, k -> new CacheAccessMetrics(new DefaultNameFactory(TYPE_NAME, String.format("%s-%s", type, k.getSimpleName()))));
}
}

73
src/java/org/apache/cassandra/metrics/CacheAccessMetrics.java Normal file
View File

@ -0,0 +1,73 @@
/*
* 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.metrics;
import com.google.common.annotations.VisibleForTesting;
import com.codahale.metrics.Meter;
import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
public class CacheAccessMetrics
{
/**
* Total number of cache hits
*/
public final Meter hits;
/**
* Total number of cache misses
*/
public final Meter misses;
/**
* Total number of cache requests
*/
public final Meter requests;
public final RatioGaugeSet hitRate;
public final RatioGaugeSet missRate;
protected final MetricNameFactory factory;
public CacheAccessMetrics(MetricNameFactory factory)
{
this.factory = factory;
this.hits = Metrics.meter(factory.createMetricName("Hits"));
this.misses = Metrics.meter(factory.createMetricName("Misses"));
this.requests = Metrics.meter(factory.createMetricName("Requests"));
this.hitRate = new RatioGaugeSet(hits, requests, factory, "%sHitRate");
this.missRate = new RatioGaugeSet(misses, requests, factory, "%sMissRate");
}
@VisibleForTesting
public void reset()
{
// No actual reset happens. The Meter counter is put to zero but will not reset the moving averages
// It rather injects a weird value into them.
// This method is being only used by CacheMetricsTest and CachingBench so fixing this issue was acknowledged
// but not considered mandatory to be fixed now (CASSANDRA-16228)
hits.mark(-hits.getCount());
misses.mark(-misses.getCount());
requests.mark(-requests.getCount());
}
}

2
src/java/org/apache/cassandra/metrics/CacheMetrics.java
View File

@ -38,7 +38,7 @@ public class CacheMetrics extends AbstractCacheMetrics
public final Gauge<Integer> entries;
/**
* Create metrics for given cache.
* Create metrics for the given cache supporting entity.
*
* @param type Type of Cache to identify metrics
* @param cache Weighted Cache to measure metrics

63
src/java/org/apache/cassandra/metrics/CacheSizeMetrics.java Normal file
View File

@ -0,0 +1,63 @@
/*
* 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.metrics;
import com.codahale.metrics.Gauge;
import org.apache.cassandra.cache.CacheSize;
import static org.apache.cassandra.metrics.CacheMetrics.TYPE_NAME;
import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
public class CacheSizeMetrics
{
public static final String CAPACITY = "Capacity";
public static final String SIZE = "Size";
public static final String ENTRIES = "Entries";
/**
* Cache capacity in bytes
*/
public final Gauge<Long> capacity;
/**
* Total size of cache, in bytes
*/
public final Gauge<Long> size;
/**
* Total number of cache entries
*/
public final Gauge<Integer> entries;
/**
* Create metrics for the given cache supporting entity.
*
* @param type Type of Cache to identify metrics.
* @param cache Cache to measure metrics
*/
public CacheSizeMetrics(String type, CacheSize cache)
{
this(new DefaultNameFactory(TYPE_NAME, type), cache);
}
public CacheSizeMetrics(MetricNameFactory factory, CacheSize cache)
{
capacity = Metrics.register(factory.createMetricName(CAPACITY), cache::capacity);
size = Metrics.register(factory.createMetricName(SIZE), cache::weightedSize);
entries = Metrics.register(factory.createMetricName(ENTRIES), cache::size);
}
}

121
src/java/org/apache/cassandra/metrics/RatioGaugeSet.java Normal file
View File

@ -0,0 +1,121 @@
/*
* 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.metrics;
import java.util.function.DoubleSupplier;
import java.util.function.ToDoubleFunction;
import com.codahale.metrics.Metered;
import com.codahale.metrics.RatioGauge;
import org.apache.cassandra.metrics.CassandraMetricsRegistry.MetricName;
import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
public class RatioGaugeSet
{
public static final String ONE_MINUTE = "OneMinute";
public static final String FIVE_MINUTE = "FiveMinute";
public static final String FIFTEEN_MINUTE = "FifteenMinute";
public static final String MEAN_RATIO = "";
public final RatioGauge oneMinute;
public final RatioGauge fiveMinute;
public final RatioGauge fifteenMinute;
public final RatioGauge mean;
public RatioGaugeSet(Metered numerator, Metered denominator, MetricNameFactory factory, String namePattern)
{
this.oneMinute = ratioGauge(factory.createMetricName(String.format(namePattern, ONE_MINUTE)), numerator::getOneMinuteRate, denominator::getOneMinuteRate);
this.fiveMinute = ratioGauge(factory.createMetricName(String.format(namePattern, FIVE_MINUTE)), numerator::getFiveMinuteRate, denominator::getFiveMinuteRate);
this.fifteenMinute = ratioGauge(factory.createMetricName(String.format(namePattern, FIFTEEN_MINUTE)), numerator::getFifteenMinuteRate, denominator::getFifteenMinuteRate);
this.mean = ratioGauge(factory.createMetricName(String.format(namePattern, MEAN_RATIO)), numerator::getCount, denominator::getCount);
}
private static RatioGauge ratioGauge(DoubleSupplier numerator, DoubleSupplier denominator)
{
return new RatioGauge()
{
protected Ratio getRatio()
{
return Ratio.of(numerator.getAsDouble(), denominator.getAsDouble());
}
};
}
private RatioGauge ratioGauge(MetricName name, DoubleSupplier numerator, DoubleSupplier denominator)
{
return Metrics.register(name, ratioGauge(numerator, denominator));
}
public static Metered sum(Metered... meters)
{
return new SummingMeter(meters);
}
private static class SummingMeter implements Metered
{
private final Metered[] meters;
public SummingMeter(Metered... meters)
{
this.meters = meters;
}
@Override
public long getCount()
{
long count = 0;
for (Metered meter : meters)
count += meter.getCount();
return count;
}
private double getRate(ToDoubleFunction<Metered> rateSupplier)
{
double rate = 0;
for (Metered meter : meters)
rate += rateSupplier.applyAsDouble(meter);
return rate;
}
@Override
public double getMeanRate()
{
return getRate(Metered::getMeanRate);
}
@Override
public double getFifteenMinuteRate()
{
return getRate(Metered::getFifteenMinuteRate);
}
@Override
public double getFiveMinuteRate()
{
return getRate(Metered::getFiveMinuteRate);
}
@Override
public double getOneMinuteRate()
{
return getRate(Metered::getOneMinuteRate);
}
}
}

44
src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
View File

@ -30,12 +30,10 @@ import java.util.concurrent.TimeUnit;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -72,10 +70,12 @@ import accord.utils.ReducingRangeMap;
import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
import accord.utils.async.Observable;
import org.apache.cassandra.cache.CacheSize;
import org.apache.cassandra.concurrent.ExecutorPlus;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.metrics.AccordStateCacheMetrics;
import org.apache.cassandra.service.accord.api.PartitionKey;
import org.apache.cassandra.service.accord.async.AsyncOperation;
import org.apache.cassandra.service.accord.async.ExecutionOrder;
@ -85,7 +85,7 @@ import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
public class AccordCommandStore extends CommandStore
public class AccordCommandStore extends CommandStore implements CacheSize
{
private static final Logger logger = LoggerFactory.getLogger(AccordCommandStore.class);
@ -124,9 +124,10 @@ public class AccordCommandStore extends CommandStore
DataStore dataStore,
ProgressLog.Factory progressLogFactory,
EpochUpdateHolder epochUpdateHolder,
AccordJournal journal)
AccordJournal journal,
AccordStateCacheMetrics cacheMetrics)
{
this(id, time, agent, dataStore, progressLogFactory, epochUpdateHolder, journal, Stage.READ.executor(), Stage.MUTATION.executor());
this(id, time, agent, dataStore, progressLogFactory, epochUpdateHolder, journal, Stage.READ.executor(), Stage.MUTATION.executor(), cacheMetrics);
}
@VisibleForTesting
@ -138,7 +139,8 @@ public class AccordCommandStore extends CommandStore
EpochUpdateHolder epochUpdateHolder,
AccordJournal journal,
ExecutorPlus loadExecutor,
ExecutorPlus saveExecutor)
ExecutorPlus saveExecutor,
AccordStateCacheMetrics cacheMetrics)
{
super(id, time, agent, dataStore, progressLogFactory, epochUpdateHolder);
this.journal = journal;
@ -146,7 +148,7 @@ public class AccordCommandStore extends CommandStore
executor = executorFactory().sequential(CommandStore.class.getSimpleName() + '[' + id + ']');
executionOrder = new ExecutionOrder();
threadId = getThreadId(executor);
stateCache = new AccordStateCache(loadExecutor, saveExecutor, 8 << 20);
stateCache = new AccordStateCache(loadExecutor, saveExecutor, 8 << 20, cacheMetrics);
commandCache =
stateCache.instance(TxnId.class,
TxnId.class,
@ -181,10 +183,10 @@ public class AccordCommandStore extends CommandStore
executor.execute(this::loadRangesToCommands);
}
static Factory factory(AccordJournal journal)
static Factory factory(AccordJournal journal, AccordStateCacheMetrics cacheMetrics)
{
return (id, time, agent, dataStore, progressLogFactory, rangesForEpoch) ->
new AccordCommandStore(id, time, agent, dataStore, progressLogFactory, rangesForEpoch, journal);
new AccordCommandStore(id, time, agent, dataStore, progressLogFactory, rangesForEpoch, journal, cacheMetrics);
}
private void loadRangesToCommands()
@ -250,15 +252,29 @@ public class AccordCommandStore extends CommandStore
return Thread.currentThread().getId() == threadId;
}
public void setCacheSize(long bytes)
@Override
public void setCapacity(long bytes)
{
checkInStoreThread();
stateCache.setMaxSize(bytes);
stateCache.setCapacity(bytes);
}
public long getCacheSize()
@Override
public long capacity()
{
return stateCache.getMaxSize();
return stateCache.capacity();
}
@Override
public int size()
{
return stateCache.size();
}
@Override
public long weightedSize()
{
return stateCache.weightedSize();
}
public void checkInStoreThread()

39
src/java/org/apache/cassandra/service/accord/AccordCommandStores.java
View File

@ -30,15 +30,24 @@ import accord.local.ShardDistributor;
import accord.primitives.Range;
import accord.topology.Topology;
import accord.utils.RandomSource;
import org.apache.cassandra.cache.CacheSize;
import org.apache.cassandra.metrics.AccordStateCacheMetrics;
import org.apache.cassandra.metrics.CacheSizeMetrics;
import org.apache.cassandra.service.accord.api.AccordRoutingKey;
public class AccordCommandStores extends CommandStores
public class AccordCommandStores extends CommandStores implements CacheSize
{
public static final String ACCORD_STATE_CACHE = "accord-state-cache";
private final CacheSizeMetrics cacheSizeMetrics;
private long cacheSize;
AccordCommandStores(NodeTimeService time, Agent agent, DataStore store, RandomSource random,
ShardDistributor shardDistributor, ProgressLog.Factory progressLogFactory, AccordJournal journal)
{
super(time, agent, store, random, shardDistributor, progressLogFactory, AccordCommandStore.factory(journal));
setCacheSize(maxCacheSize());
super(time, agent, store, random, shardDistributor, progressLogFactory, AccordCommandStore.factory(journal, new AccordStateCacheMetrics(ACCORD_STATE_CACHE)));
setCapacity(maxCacheSize());
this.cacheSizeMetrics = new CacheSizeMetrics(ACCORD_STATE_CACHE, this);
}
static Factory factory(AccordJournal journal)
@ -67,21 +76,37 @@ public class AccordCommandStores extends CommandStores
return false;
}
private long cacheSize;
synchronized void setCacheSize(long bytes)
public synchronized void setCapacity(long bytes)
{
cacheSize = bytes;
refreshCacheSizes();
}
@Override
public long capacity()
{
return cacheSize;
}
@Override
public int size()
{
return unsafeFoldLeft(0, (size, commandStore) -> size + ((AccordCommandStore) commandStore).size());
}
@Override
public long weightedSize()
{
return unsafeFoldLeft(0L, (size, commandStore) -> size + ((AccordCommandStore) commandStore).weightedSize());
}
synchronized void refreshCacheSizes()
{
if (count() == 0)
return;
long perStore = cacheSize / count();
// TODO (low priority, safety): we might transiently breach our limit if we increase one store before decreasing another
forEach(commandStore -> ((AccordSafeCommandStore) commandStore).commandStore().setCacheSize(perStore));
forEach(commandStore -> ((AccordSafeCommandStore) commandStore).commandStore().setCapacity(perStore));
}
private static long maxCacheSize()

3
src/java/org/apache/cassandra/service/accord/AccordService.java
View File

@ -299,7 +299,6 @@ public class AccordService implements IAccordService, Shutdownable
}
if (cause instanceof Preempted)
{
metrics.preempts.mark();
//TODO need to improve
// Coordinator "could" query the accord state to see whats going on but that doesn't exist yet.
// Protocol also doesn't have a way to denote "unknown" outcome, so using a timeout as the closest match
@ -372,7 +371,7 @@ public class AccordService implements IAccordService, Shutdownable
{
long bytes = kb << 10;
AccordCommandStores commandStores = (AccordCommandStores) node.commandStores();
commandStores.setCacheSize(bytes);
commandStores.setCapacity(bytes);
}
@Override

133
src/java/org/apache/cassandra/service/accord/AccordStateCache.java
View File

@ -26,13 +26,15 @@ import java.util.function.ToLongFunction;
import java.util.stream.Stream;
import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.utils.IntrusiveLinkedList;
import accord.utils.async.AsyncChains;
import org.apache.cassandra.cache.CacheSize;
import org.apache.cassandra.concurrent.ExecutorPlus;
import org.apache.cassandra.metrics.AccordStateCacheMetrics;
import org.apache.cassandra.metrics.CacheAccessMetrics;
import org.apache.cassandra.service.accord.AccordCachingState.Status;
import static accord.utils.Invariants.checkState;
@ -49,7 +51,7 @@ import static org.apache.cassandra.service.accord.AccordCachingState.Status.SAVI
* Supports dynamic object sizes. After each acquire/free cycle, the cacheable objects size is recomputed to
* account for data added/removed during txn processing if it's modified flag is set
*/
public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?>>
public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?>> implements CacheSize
{
private static final Logger logger = LoggerFactory.getLogger(AccordStateCache.class);
@ -63,13 +65,6 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
VALIDATE_LOAD_ON_EVICT = value;
}
static class Stats
{
private long queries;
private long hits;
private long misses;
}
private final Map<Object, AccordCachingState<?, ?>> cache = new HashMap<>();
private final HashMap<Class<?>, Instance<?, ?, ?>> instances = new HashMap<>();
@ -78,22 +73,27 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
private int unreferenced = 0;
private long maxSizeInBytes;
private long bytesCached = 0;
private final Stats stats = new Stats();
public AccordStateCache(ExecutorPlus loadExecutor, ExecutorPlus saveExecutor, long maxSizeInBytes)
@VisibleForTesting
final AccordStateCacheMetrics metrics;
public AccordStateCache(ExecutorPlus loadExecutor, ExecutorPlus saveExecutor, long maxSizeInBytes, AccordStateCacheMetrics metrics)
{
this.loadExecutor = loadExecutor;
this.saveExecutor = saveExecutor;
this.maxSizeInBytes = maxSizeInBytes;
this.metrics = metrics;
}
public void setMaxSize(long size)
@Override
public void setCapacity(long sizeInBytes)
{
maxSizeInBytes = size;
maxSizeInBytes = sizeInBytes;
maybeEvictSomeNodes();
}
public long getMaxSize()
@Override
public long capacity()
{
return maxSizeInBytes;
}
@ -114,7 +114,11 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
private <K, V> void maybeUpdateSize(AccordCachingState<?, ?> node, ToLongFunction<?> estimator)
{
if (node.shouldUpdateSize())
bytesCached += ((AccordCachingState<K, V>) node).estimatedSizeOnHeapDelta((ToLongFunction<V>) estimator);
{
long delta = ((AccordCachingState<K, V>) node).estimatedSizeOnHeapDelta((ToLongFunction<V>) estimator);
bytesCached += delta;
instanceForNode(node).bytesCached += delta;
}
}
/*
@ -174,6 +178,8 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
checkState(!isInQueue(node));
bytesCached -= node.lastQueriedEstimatedSizeOnHeap;
Instance<?, ?, ?> instance = instanceForNode(node);
instance.bytesCached -= node.lastQueriedEstimatedSizeOnHeap;
if (node.status() == LOADED && VALIDATE_LOAD_ON_EVICT)
instanceForNode(node).validateLoadEvicted(node);
@ -181,6 +187,8 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
if (!node.hasListeners())
{
AccordCachingState<?, ?> self = cache.remove(node.key());
if (self != null)
instance.itemsCached--;
checkState(self == node, "Leaked node detected; was attempting to remove %s but cache had %s", node, self);
}
else
@ -212,7 +220,7 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
return instance;
}
public class Instance<K, V, S extends AccordSafeState<K, V>>
public class Instance<K, V, S extends AccordSafeState<K, V>> implements CacheSize
{
private final Class<K> keyClass;
private final Function<AccordCachingState<K, V>, S> safeRefFactory;
@ -220,7 +228,11 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
private BiFunction<V, V, Runnable> saveFunction;
private final BiFunction<K, V, Boolean> validateFunction;
private final ToLongFunction<V> heapEstimator;
private final Stats stats = new Stats();
private long bytesCached;
private int itemsCached;
@VisibleForTesting
final CacheAccessMetrics instanceMetrics;
public Instance(
Class<K> keyClass,
@ -236,6 +248,7 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
this.saveFunction = saveFunction;
this.validateFunction = validateFunction;
this.heapEstimator = heapEstimator;
this.instanceMetrics = metrics.forInstance(keyClass);
}
public Stream<AccordCachingState<K, V>> stream()
@ -280,8 +293,11 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
AccordCachingState<K, V> node = new AccordCachingState<>(key);
node.load(loadExecutor, loadFunction);
node.references++;
cache.put(key, node);
if (cache.put(key, node) == null)
itemsCached++;
maybeUpdateSize(node, heapEstimator);
metrics.objectSize.update(node.lastQueriedEstimatedSizeOnHeap);
maybeEvictSomeNodes();
return node;
}
@ -413,37 +429,22 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
node.complete();
}
public long cacheQueries()
{
return stats.queries;
}
public long cacheHits()
{
return stats.hits;
}
public long cacheMisses()
{
return stats.misses;
}
private void incrementCacheQueries()
{
stats.queries++;
AccordStateCache.this.stats.queries++;
instanceMetrics.requests.mark();
metrics.requests.mark();
}
private void incrementCacheHits()
{
stats.hits++;
AccordStateCache.this.stats.hits++;
instanceMetrics.hits.mark();
metrics.hits.mark();
}
private void incrementCacheMisses()
{
stats.misses++;
AccordStateCache.this.stats.misses++;
instanceMetrics.misses.mark();
metrics.misses.mark();
}
@VisibleForTesting
@ -457,12 +458,43 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
{
this.saveFunction = saveFunction;
}
@Override
public long capacity()
{
return AccordStateCache.this.capacity();
}
@Override
public void setCapacity(long capacity)
{
throw new UnsupportedOperationException("Capacity is shared between all instances. Please set the capacity on the global cache");
}
@Override
public int size()
{
return itemsCached;
}
@Override
public long weightedSize()
{
return bytesCached;
}
}
@VisibleForTesting
void unsafeClear()
{
cache.clear();
bytesCached = 0;
metrics.reset();;
instances.values().forEach(i -> {
i.itemsCached = 0;
i.bytesCached = 0;
i.instanceMetrics.reset();
});
//noinspection StatementWithEmptyBody
while (null != poll());
}
@ -504,14 +536,14 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
return unreferenced;
}
@VisibleForTesting
int totalNumEntries()
@Override
public int size()
{
return cache.size();
}
@VisibleForTesting
long bytesCached()
@Override
public long weightedSize()
{
return bytesCached;
}
@ -536,19 +568,4 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
AccordCachingState<?, ?> node = cache.get(key);
return node != null ? node.references : 0;
}
public long cacheQueries()
{
return stats.queries;
}
public long cacheHits()
{
return stats.hits;
}
public long cacheMisses()
{
return stats.misses;
}
}

8
src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
View File

@ -24,6 +24,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.Agent;
import accord.api.EventsListener;
import accord.api.Result;
import accord.local.Command;
import accord.local.Node;
@ -33,6 +34,7 @@ import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.metrics.AccordMetrics;
import org.apache.cassandra.service.accord.txn.TxnQuery;
import org.apache.cassandra.service.accord.txn.TxnRead;
import org.apache.cassandra.utils.JVMStabilityInspector;
@ -101,4 +103,10 @@ public class AccordAgent implements Agent
{
return new Txn.InMemory(kind, keysOrRanges, TxnRead.EMPTY, TxnQuery.ALL, null);
}
@Override
public EventsListener metricsEventsListener()
{
return AccordMetrics.Listener.instance;
}
}

273
test/distributed/org/apache/cassandra/distributed/test/accord/AccordMetricsTest.java Normal file
View File

@ -0,0 +1,273 @@
/*
* 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.accord;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Function;
import com.google.common.base.Throwables;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.IMessageFilters;
import org.apache.cassandra.exceptions.ReadTimeoutException;
import org.apache.cassandra.exceptions.WriteTimeoutException;
import org.apache.cassandra.metrics.AccordMetrics;
import org.apache.cassandra.metrics.DefaultNameFactory;
import org.apache.cassandra.metrics.RatioGaugeSet;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.exceptions.ReadPreemptedException;
import org.apache.cassandra.service.accord.exceptions.WritePreemptedException;
import org.assertj.core.data.Offset;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
public class AccordMetricsTest extends AccordTestBase
{
private static final Logger logger = LoggerFactory.getLogger(AccordMetricsTest.class);
@Override
protected Logger logger()
{
return logger;
}
@BeforeClass
public static void setupClass() throws IOException
{
AccordTestBase.setupClass();
SHARED_CLUSTER.forEach(node -> node.runOnInstance(() -> AccordService.instance().setCacheSize(0)));
for (int i = 0; i < SHARED_CLUSTER.size(); i++) // initialize metrics
logger.trace(SHARED_CLUSTER.get(i + 1).callOnInstance(() -> AccordMetrics.readMetrics.toString() + AccordMetrics.writeMetrics.toString()));
}
String writeCql()
{
return "BEGIN TRANSACTION\n" +
" LET val = (SELECT v FROM " + currentTable + " WHERE k=? AND c=?);\n" +
" SELECT val.v;\n" +
" UPDATE " + currentTable + " SET v = v + 1 WHERE k=? AND c=?;\n" +
"COMMIT TRANSACTION";
}
String readCql()
{
return "BEGIN TRANSACTION\n" +
" LET val = (SELECT v FROM " + currentTable + " WHERE k=? AND c=?);\n" +
" SELECT val.v;\n" +
"COMMIT TRANSACTION";
}
Map<Integer, Map<String, Long>> countingMetrics0;
@Before
public void beforeTest()
{
SHARED_CLUSTER.filters().reset();
SHARED_CLUSTER.schemaChange("CREATE TABLE " + currentTable + " (k int, c int, v int, PRIMARY KEY (k, c))");
SHARED_CLUSTER.coordinator(1).execute("INSERT INTO " + currentTable + " (k, c, v) VALUES (0, 0, 0)", ConsistencyLevel.ALL);
}
@Test
public void testRegularMetrics() throws Exception
{
countingMetrics0 = getMetrics();
SHARED_CLUSTER.coordinator(1).executeWithResult(writeCql(), ConsistencyLevel.ALL, 0, 0, 0, 0);
assertCoordinatorMetrics(0, "rw", 1, 0, 0, 0, 0);
assertCoordinatorMetrics(1, "rw", 0, 0, 0, 0, 0);
assertReplicaMetrics(0, "rw", 1, 1, 1);
assertReplicaMetrics(1, "rw", 1, 1, 1);
assertZeroMetrics("ro");
countingMetrics0 = getMetrics();
SHARED_CLUSTER.coordinator(1).executeWithResult(readCql(), ConsistencyLevel.ALL, 0, 0);
assertCoordinatorMetrics(0, "ro", 1, 0, 0, 0, 0);
assertCoordinatorMetrics(1, "ro", 0, 0, 0, 0, 0);
assertReplicaMetrics(0, "ro", 1, 1, 0);
assertReplicaMetrics(1, "ro", 1, 1, 0);
assertZeroMetrics("rw");
}
@Test
public void testPreemptionMetrics()
{
IMessageFilters.Filter commitFilter1 = SHARED_CLUSTER.filters().outbound().verbs(Verb.ACCORD_COMMIT_REQ.id).from(1).to(1).drop();
IMessageFilters.Filter commitFilter2 = SHARED_CLUSTER.filters().outbound().verbs(Verb.ACCORD_COMMIT_REQ.id).from(1).to(2).drop();
commitFilter1.on();
commitFilter2.on();
countingMetrics0 = getMetrics();
try
{
SHARED_CLUSTER.coordinator(1).executeWithResult(writeCql(), ConsistencyLevel.ALL, 0, 0, 0, 0);
fail("expected to fail");
}
catch (RuntimeException ex)
{
assertThat(Throwables.getCausalChain(ex).stream().map(t -> t.getClass().getName())).contains(WritePreemptedException.class.getName());
}
assertCoordinatorMetrics(0, "rw", 1, 0, 1, 0, 0);
assertCoordinatorMetrics(1, "rw", 0, 0, 0, 0, 1);
assertReplicaMetrics(0, "rw", 1, 1, 1);
assertReplicaMetrics(1, "rw", 1, 1, 1);
assertZeroMetrics("ro");
countingMetrics0 = getMetrics();
try
{
SHARED_CLUSTER.coordinator(1).executeWithResult(readCql(), ConsistencyLevel.ALL, 0, 0);
fail("expected to fail");
}
catch (RuntimeException ex)
{
assertThat(Throwables.getCausalChain(ex).stream().map(t -> t.getClass().getName())).contains(ReadPreemptedException.class.getName());
}
assertCoordinatorMetrics(0, "ro", 1, 0, 1, 0, 0);
assertCoordinatorMetrics(1, "ro", 0, 0, 0, 0, 1);
assertReplicaMetrics(0, "ro", 1, 1, 0);
assertReplicaMetrics(1, "ro", 1, 1, 0);
assertZeroMetrics("rw");
}
@Test
public void testTimeoutMetrics()
{
IMessageFilters.Filter preAcceptFilter = SHARED_CLUSTER.filters().outbound().verbs(Verb.ACCORD_PRE_ACCEPT_REQ.id).from(1).to(2).drop();
preAcceptFilter.on();
countingMetrics0 = getMetrics();
try
{
SHARED_CLUSTER.coordinator(1).executeWithResult(readCql(), ConsistencyLevel.ALL, 0, 0);
fail("expected to fail");
}
catch (RuntimeException ex)
{
assertThat(Throwables.getCausalChain(ex).stream().map(t -> t.getClass().getName())).contains(ReadTimeoutException.class.getName());
}
assertCoordinatorMetrics(0, "ro", 0, 0, 0, 1, 0);
assertCoordinatorMetrics(1, "ro", 0, 0, 0, 0, 0);
assertReplicaMetrics(0, "ro", 0, 0, 0);
assertReplicaMetrics(1, "ro", 0, 0, 0);
assertZeroMetrics("rw");
countingMetrics0 = getMetrics();
try
{
SHARED_CLUSTER.coordinator(1).executeWithResult(writeCql(), ConsistencyLevel.ALL, 0, 0, 0, 0);
fail("expected to fail");
}
catch (RuntimeException ex)
{
assertThat(Throwables.getCausalChain(ex).stream().map(t -> t.getClass().getName())).contains(WriteTimeoutException.class.getName());
}
assertCoordinatorMetrics(0, "rw", 0, 0, 0, 1, 0);
assertCoordinatorMetrics(1, "rw", 0, 0, 0, 0, 0);
assertReplicaMetrics(0, "rw", 0, 0, 0);
assertReplicaMetrics(1, "rw", 0, 0, 0);
assertZeroMetrics("ro");
}
private void assertZeroMetrics(String scope)
{
for (int i = 0; i < SHARED_CLUSTER.size(); i++)
{
assertCoordinatorMetrics(i, scope, 0, 0, 0, 0, 0);
assertReplicaMetrics(i, scope, 0, 0, 0);
}
}
private void assertCoordinatorMetrics(int node, String scope, long fastPaths, long slowPaths, long preempts, long timeouts, long recoveries)
{
DefaultNameFactory nameFactory = new DefaultNameFactory(AccordMetrics.ACCORD_COORDINATOR, scope);
Map<String, Long> metrics = diff(countingMetrics0).get(node);
logger.info("Metrics for node {} / {}: {}", node, scope, metrics);
Function<String, Long> metric = n -> metrics.get(nameFactory.createMetricName(n).getMetricName());
assertThat(metric.apply(AccordMetrics.FAST_PATHS)).isEqualTo(fastPaths);
assertThat(metric.apply(AccordMetrics.SLOW_PATHS)).isEqualTo(slowPaths);
assertThat(metric.apply(AccordMetrics.PREEMPTS)).isEqualTo(preempts);
assertThat(metric.apply(AccordMetrics.TIMEOUTS)).isEqualTo(timeouts);
assertThat(metric.apply(AccordMetrics.RECOVERY_DELAY)).isEqualTo(recoveries);
assertThat(metric.apply(AccordMetrics.RECOVERY_TIME)).isEqualTo(recoveries);
assertThat(metric.apply(AccordMetrics.DEPENDENCIES)).isEqualTo(fastPaths + slowPaths);
if ((fastPaths + slowPaths) > 0)
{
String fastPathToTotalName = nameFactory.createMetricName(AccordMetrics.FAST_PATH_TO_TOTAL + "." + RatioGaugeSet.MEAN_RATIO).getMetricName();
assertThat((double) SHARED_CLUSTER.get(1).metrics().getGauge(fastPathToTotalName)).isEqualTo((double) fastPaths / (double) (fastPaths + slowPaths), Offset.offset(0.01d));
}
}
private void assertReplicaMetrics(int node, String scope, long commits, long executions, long applications)
{
DefaultNameFactory nameFactory = new DefaultNameFactory(AccordMetrics.ACCORD_REPLICA, scope);
Map<String, Long> metrics = diff(countingMetrics0).get(node);
Function<String, Long> metric = n -> metrics.get(nameFactory.createMetricName(n).getMetricName());
assertThat(metric.apply(AccordMetrics.COMMIT_LATENCY)).isEqualTo(commits);
assertThat(metric.apply(AccordMetrics.EXECUTE_LATENCY)).isEqualTo(executions);
assertThat(metric.apply(AccordMetrics.APPLY_LATENCY)).isEqualTo(applications);
assertThat(metric.apply(AccordMetrics.APPLY_DURATION)).isEqualTo(applications);
assertThat(metric.apply(AccordMetrics.PARTIAL_DEPENDENCIES)).isEqualTo(executions);
}
private Map<Integer, Map<String, Long>> getMetrics()
{
Map<Integer, Map<String, Long>> metrics = new HashMap<>();
for (int i = 0; i < SHARED_CLUSTER.size(); i++)
metrics.put(i, SHARED_CLUSTER.get(i + 1).metrics().getCounters(name -> name.startsWith("org.apache.cassandra.metrics.accord-")));
return metrics;
}
private Map<Integer, Map<String, Long>> diff(Map<Integer, Map<String, Long>> prev)
{
Map<Integer, Map<String, Long>> curr = getMetrics();
Map<Integer, Map<String, Long>> diff = new HashMap<>();
for (int i = 0; i < SHARED_CLUSTER.size(); i++)
{
Map<String, Long> prevNode = prev.get(i);
Map<String, Long> currNode = curr.get(i);
Map<String, Long> diffNode = new HashMap<>();
for (Map.Entry<String, Long> currEntry : currNode.entrySet())
{
Long prevVal = prevNode.get(currEntry.getKey());
if (prevVal != null)
diffNode.put(currEntry.getKey(), currEntry.getValue() - prevVal);
}
diff.put(i, diffNode);
}
return diff;
}
}

6
test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java
View File

@ -399,9 +399,9 @@ public class CompactionAccordIteratorsTest
{
commandStore.executeBlocking(() -> {
// clear cache and wait for post-eviction writes to complete
long cacheSize = commandStore.getCacheSize();
commandStore.setCacheSize(0);
commandStore.setCacheSize(cacheSize);
long cacheSize = commandStore.capacity();
commandStore.setCapacity(0);
commandStore.setCapacity(cacheSize);
commandStore.cache().awaitSaveResults();
});
commands.forceBlockingFlush(FlushReason.UNIT_TESTS);

6
test/unit/org/apache/cassandra/io/sstable/LargePartitionsTest.java
View File

@ -132,10 +132,10 @@ public class LargePartitionsTest extends CQLTester
{
CacheMetrics metrics = CacheService.instance.keyCache.getMetrics();
System.out.println("Key cache metrics " + title + ": capacity:" + metrics.capacity.getValue() +
" size:"+metrics.size.getValue()+
" size:" + metrics.size.getValue() +
" entries:" + metrics.entries.getValue() +
" hit-rate:"+metrics.hitRate.getValue() +
" one-min-rate:"+metrics.oneMinuteHitRate.getValue());
" hit-rate:" + metrics.hitRate.getValue() +
" one-min-rate:" + metrics.hitRate.getValue());
}
@Test

4
test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java
View File

@ -89,7 +89,7 @@ public class AccordCommandTest
public void basicCycleTest() throws Throwable
{
AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl");
getUninterruptibly(commandStore.execute(PreLoadContext.empty(), unused -> commandStore.setCacheSize(0)));
getUninterruptibly(commandStore.execute(PreLoadContext.empty(), unused -> commandStore.setCapacity(0)));
TxnId txnId = txnId(1, clock.incrementAndGet(), 1);
Txn txn = createWriteTxn(1);
@ -173,7 +173,7 @@ public class AccordCommandTest
public void computeDeps() throws Throwable
{
AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl");
getUninterruptibly(commandStore.execute(PreLoadContext.empty(), unused -> commandStore.setCacheSize(0)));
getUninterruptibly(commandStore.execute(PreLoadContext.empty(), unused -> commandStore.setCapacity(0)));
TxnId txnId1 = txnId(1, clock.incrementAndGet(), 1);
Txn txn = createWriteTxn(2);

179
test/unit/org/apache/cassandra/service/accord/AccordStateCacheTest.java
View File

@ -17,55 +17,66 @@
*/
package org.apache.cassandra.service.accord;
import java.util.UUID;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import accord.utils.async.AsyncChain;
import org.apache.cassandra.cache.CacheSize;
import org.apache.cassandra.concurrent.ManualExecutor;
import org.apache.cassandra.metrics.AccordStateCacheMetrics;
import org.apache.cassandra.metrics.CacheAccessMetrics;
import org.apache.cassandra.service.accord.AccordCachingState.Status;
import static org.apache.cassandra.service.accord.AccordTestUtils.testLoad;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
public class AccordStateCacheTest
{
private static final long DEFAULT_NODE_SIZE = nodeSize(0);
private AccordStateCacheMetrics cacheMetrics;
private static class SafeString implements AccordSafeState<String, String>
private static abstract class TestSafeState<T> implements AccordSafeState<T, T>
{
private boolean invalidated = false;
private final AccordCachingState<String, String> global;
private String original = null;
protected boolean invalidated = false;
protected final AccordCachingState<T, T> global;
private T original = null;
public SafeString(AccordCachingState<String, String> global)
public TestSafeState(AccordCachingState<T, T> global)
{
this.global = global;
}
public AccordCachingState<String, String> global()
public AccordCachingState<T, T> global()
{
return global;
}
@Override
public String key()
public T key()
{
return global.key();
}
@Override
public String current()
public T current()
{
return global.get();
}
@Override
public void set(String update)
public void set(T update)
{
global.set(update);
}
@Override
public String original()
public T original()
{
return original;
}
@ -118,6 +129,22 @@ public class AccordStateCacheTest
}
}
private static class SafeString extends TestSafeState<String>
{
public SafeString(AccordCachingState<String, String> global)
{
super(global);
}
}
private static class SafeInt extends TestSafeState<Integer>
{
public SafeInt(AccordCachingState<Integer, Integer> global)
{
super(global);
}
}
private static long emptyNodeSize()
{
return AccordCachingState.EMPTY_SIZE;
@ -131,15 +158,35 @@ public class AccordStateCacheTest
private static void assertCacheState(AccordStateCache cache, int referenced, int total, long bytes)
{
Assert.assertEquals(referenced, cache.numReferencedEntries());
Assert.assertEquals(total, cache.totalNumEntries());
Assert.assertEquals(bytes, cache.bytesCached());
Assert.assertEquals(total, cache.size());
Assert.assertEquals(bytes, cache.weightedSize());
}
private void assertCacheMetrics(CacheAccessMetrics metrics, int hits, int misses, int requests)
{
Assert.assertEquals(hits, metrics.hits.getCount());
Assert.assertEquals(misses, metrics.misses.getCount());
Assert.assertEquals(requests, metrics.requests.getCount());
if (metrics instanceof AccordStateCacheMetrics)
{
AccordStateCacheMetrics ascMetrics = (AccordStateCacheMetrics) metrics;
Assert.assertEquals(misses, ascMetrics.objectSize.getCount());
assertThat(ascMetrics.objectSize.getSnapshot().getMax()).isGreaterThanOrEqualTo(DEFAULT_NODE_SIZE);
}
}
@Before
public void before()
{
String type = String.format("%s-%s", AccordCommandStores.ACCORD_STATE_CACHE, UUID.randomUUID());
cacheMetrics = new AccordStateCacheMetrics(type);
}
@Test
public void testAcquisitionAndRelease()
{
ManualExecutor executor = new ManualExecutor();
AccordStateCache cache = new AccordStateCache(executor, executor,500);
AccordStateCache cache = new AccordStateCache(executor, executor, 500, cacheMetrics);
AccordStateCache.Instance<String, String, SafeString> instance =
cache.instance(String.class, String.class, SafeString::new, key -> key, (original, current) -> null, (k, v) -> true, String::length);
assertCacheState(cache, 0, 0, 0);
@ -162,13 +209,56 @@ public class AccordStateCacheTest
Assert.assertSame(safeString1.global, cache.head());
Assert.assertSame(safeString2.global, cache.tail());
assertCacheMetrics(cache.metrics, 0, 2, 2);
assertCacheMetrics(instance.instanceMetrics, 0, 2, 2);
}
@Test
public void testCachingMetricsWithTwoInstances()
{
ManualExecutor executor = new ManualExecutor();
AccordStateCache cache = new AccordStateCache(executor, executor, 500, cacheMetrics);
AccordStateCache.Instance<String, String, SafeString> stringInstance =
cache.instance(String.class, String.class, SafeString::new, key -> key, (original, current) -> null, (k, v) -> true,String::length);
AccordStateCache.Instance<Integer, Integer, SafeInt> intInstance =
cache.instance(Integer.class, Integer.class, SafeInt::new, key -> key, (original, current) -> null, (k, v) -> true,ignored -> Integer.BYTES);
assertCacheState(cache, 0, 0, 0);
SafeString safeString1 = stringInstance.acquire("1");
testLoad(executor, safeString1, "1");
stringInstance.release(safeString1);
SafeString safeString2 = stringInstance.acquire("2");
testLoad(executor, safeString2, "2");
stringInstance.release(safeString2);
SafeInt safeInt1 = intInstance.acquire(3);
testLoad(executor, safeInt1, 3);
intInstance.release(safeInt1);
SafeInt safeInt2 = intInstance.acquire(4);
testLoad(executor, safeInt2, 4);
intInstance.release(safeInt2);
SafeInt safeInt3 = intInstance.acquire(5);
testLoad(executor, safeInt3, 5);
intInstance.release(safeInt3);
assertCacheState(cache, 0, 5, nodeSize(Integer.BYTES) * 3 + nodeSize(1) * 2);
assertThat(stringInstance.size()).isEqualTo(2);
assertThat(stringInstance.weightedSize()).isEqualTo(nodeSize(1) * 2);
assertThat(stringInstance.capacity()).isEqualTo(cache.capacity());
assertThat(intInstance.size()).isEqualTo(3);
assertThat(intInstance.weightedSize()).isEqualTo(nodeSize(Integer.BYTES) * 3);
assertThat(intInstance.capacity()).isEqualTo(cache.capacity());
assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(() -> stringInstance.setCapacity(123));
assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(() -> intInstance.setCapacity(123));
}
@Test
public void testRotation()
{
ManualExecutor executor = new ManualExecutor();
AccordStateCache cache = new AccordStateCache(executor, executor, DEFAULT_NODE_SIZE * 5);
AccordStateCache cache = new AccordStateCache(executor, executor, DEFAULT_NODE_SIZE * 5, cacheMetrics);
AccordStateCache.Instance<String, String, SafeString> instance =
cache.instance(String.class, String.class, SafeString::new, key -> key, (original, current) -> null, (k, v) -> true, String::length);
assertCacheState(cache, 0, 0, 0);
@ -187,11 +277,15 @@ public class AccordStateCacheTest
Assert.assertSame(items[0].global, cache.head());
Assert.assertSame(items[2].global, cache.tail());
assertCacheState(cache, 0, 3, nodeSize(1) * 3);
assertCacheMetrics(cache.metrics, 0, 3, 3);
assertCacheMetrics(instance.instanceMetrics, 0, 3, 3);
SafeString safeString = instance.acquire("1");
Assert.assertEquals(Status.LOADED, safeString.globalStatus());
assertCacheState(cache, 1, 3, nodeSize(1) * 3);
assertCacheMetrics(cache.metrics, 1, 3, 4);
assertCacheMetrics(instance.instanceMetrics, 1, 3, 4);
// releasing item should return it to the tail
instance.release(safeString);
@ -204,7 +298,7 @@ public class AccordStateCacheTest
public void testEvictionOnAcquire()
{
ManualExecutor executor = new ManualExecutor();
AccordStateCache cache = new AccordStateCache(executor, executor, nodeSize(1) * 5);
AccordStateCache cache = new AccordStateCache(executor, executor, nodeSize(1) * 5, cacheMetrics);
AccordStateCache.Instance<String, String, SafeString> instance =
cache.instance(String.class, String.class, SafeString::new, key -> key, (original, current) -> null, (k, v) -> true, String::length);
assertCacheState(cache, 0, 0, 0);
@ -222,6 +316,8 @@ public class AccordStateCacheTest
assertCacheState(cache, 0, 5, nodeSize(1) * 5);
Assert.assertSame(items[0].global, cache.head());
Assert.assertSame(items[4].global, cache.tail());
assertCacheMetrics(cache.metrics, 0, 5, 5);
assertCacheMetrics(instance.instanceMetrics, 0, 5, 5);
SafeString safeString = instance.acquire("5");
Assert.assertTrue(instance.isReferenced(safeString.key()));
@ -232,19 +328,23 @@ public class AccordStateCacheTest
Assert.assertSame(items[4].global, cache.tail());
Assert.assertFalse(cache.keyIsCached("0"));
Assert.assertFalse(cache.keyIsReferenced("0"));
assertCacheMetrics(cache.metrics, 0, 6, 6);
assertCacheMetrics(instance.instanceMetrics, 0, 6, 6);
testLoad(executor, safeString, "5");
instance.release(safeString);
assertCacheState(cache, 0, 5, nodeSize(1) * 5);
Assert.assertSame(items[1].global, cache.head());
Assert.assertSame(safeString.global, cache.tail());
assertCacheMetrics(cache.metrics, 0, 6, 6);
assertCacheMetrics(instance.instanceMetrics, 0, 6, 6);
}
@Test
public void testEvictionOnRelease()
{
ManualExecutor executor = new ManualExecutor();
AccordStateCache cache = new AccordStateCache(executor, executor, nodeSize(1) * 4);
AccordStateCache cache = new AccordStateCache(executor, executor, nodeSize(1) * 4, cacheMetrics);
AccordStateCache.Instance<String, String, SafeString> instance =
cache.instance(String.class, String.class, SafeString::new, key -> key, (original, current) -> null, (k, v) -> true, String::length);
assertCacheState(cache, 0, 0, 0);
@ -259,16 +359,22 @@ public class AccordStateCacheTest
}
assertCacheState(cache, 5, 5, nodeSize(0) * 5);
assertCacheMetrics(cache.metrics, 0, 5, 5);
assertCacheMetrics(instance.instanceMetrics, 0, 5, 5);
Assert.assertNull(cache.head());
Assert.assertNull(cache.tail());
instance.release(items[2]);
assertCacheState(cache, 4, 4, nodeSize(0) * 4);
assertCacheMetrics(cache.metrics, 0, 5, 5);
assertCacheMetrics(instance.instanceMetrics, 0, 5, 5);
Assert.assertNull(cache.head());
Assert.assertNull(cache.tail());
instance.release(items[4]);
assertCacheState(cache, 3, 4, nodeSize(0) * 3 + nodeSize(1));
assertCacheMetrics(cache.metrics, 0, 5, 5);
assertCacheMetrics(instance.instanceMetrics, 0, 5, 5);
Assert.assertSame(items[4].global, cache.head());
Assert.assertSame(items[4].global, cache.tail());
}
@ -277,7 +383,7 @@ public class AccordStateCacheTest
public void testMultiAcquireRelease()
{
ManualExecutor executor = new ManualExecutor();
AccordStateCache cache = new AccordStateCache(executor, executor, DEFAULT_NODE_SIZE * 4);
AccordStateCache cache = new AccordStateCache(executor, executor, DEFAULT_NODE_SIZE * 4, cacheMetrics);
AccordStateCache.Instance<String, String, SafeString> instance =
cache.instance(String.class, String.class, SafeString::new, key -> key, (original, current) -> null, (k, v) -> true, String::length);
assertCacheState(cache, 0, 0, 0);
@ -285,6 +391,8 @@ public class AccordStateCacheTest
SafeString safeString1 = instance.acquire("0");
testLoad(executor, safeString1, "0");
Assert.assertEquals(Status.LOADED, safeString1.globalStatus());
assertCacheMetrics(cache.metrics, 0, 1, 1);
assertCacheMetrics(instance.instanceMetrics, 0, 1, 1);
Assert.assertEquals(1, cache.references("0"));
assertCacheState(cache, 1, 1, nodeSize(0));
@ -293,6 +401,8 @@ public class AccordStateCacheTest
Assert.assertEquals(Status.LOADED, safeString1.globalStatus());
Assert.assertEquals(2, cache.references("0"));
assertCacheState(cache, 1, 1, nodeSize(0));
assertCacheMetrics(cache.metrics, 1, 1, 2);
assertCacheMetrics(instance.instanceMetrics, 1, 1, 2);
instance.release(safeString1);
assertCacheState(cache, 1, 1, nodeSize(1));
@ -304,7 +414,7 @@ public class AccordStateCacheTest
public void evictionBlockedOnSaving()
{
ManualExecutor executor = new ManualExecutor();
AccordStateCache cache = new AccordStateCache(executor, executor, nodeSize(1) * 3 + nodeSize(3));
AccordStateCache cache = new AccordStateCache(executor, executor, nodeSize(1) * 3 + nodeSize(3), cacheMetrics);
AccordStateCache.Instance<String, String, SafeString> instance =
cache.instance(String.class, String.class, SafeString::new, key -> key, (original, current) -> null, (k, v) -> true, String::length);
assertCacheState(cache, 0, 0, 0);
@ -324,9 +434,11 @@ public class AccordStateCacheTest
}
assertCacheState(cache, 0, 4, nodeSize(1) * 3 + nodeSize(3));
assertCacheMetrics(cache.metrics, 0, 4, 4);
assertCacheMetrics(instance.instanceMetrics, 0, 4, 4);
// force cache eviction
cache.setMaxSize(0);
cache.setCapacity(0);
// all should have been evicted except 0
assertCacheState(cache, 0, 1, nodeSize(2));
@ -341,7 +453,7 @@ public class AccordStateCacheTest
public void testUpdates()
{
ManualExecutor executor = new ManualExecutor();
AccordStateCache cache = new AccordStateCache(executor, executor, 500);
AccordStateCache cache = new AccordStateCache(executor, executor, 500, cacheMetrics);
AccordStateCache.Instance<String, String, SafeString> instance =
cache.instance(String.class, String.class, SafeString::new, key -> key, (original, current) -> null, (k, v) -> true, String::length);
assertCacheState(cache, 0, 0, 0);
@ -360,5 +472,32 @@ public class AccordStateCacheTest
assertCacheState(cache, 0, 1, nodeSize(3));
Assert.assertSame(safeString.global, cache.head());
Assert.assertSame(safeString.global, cache.tail());
assertCacheMetrics(cache.metrics, 0, 1, 1);
assertCacheMetrics(instance.instanceMetrics, 0, 1, 1);
}
private CacheSize mockCacheSize(long capacity, long size, int entries)
{
CacheSize cacheSize = mock(CacheSize.class);
when(cacheSize.capacity()).thenReturn(capacity);
when(cacheSize.weightedSize()).thenReturn(size);
when(cacheSize.size()).thenReturn(entries);
return cacheSize;
}
@Test
public void testAccorStateCacheMetrics()
{
CacheAccessMetrics stringInstance1 = cacheMetrics.forInstance(String.class);
CacheAccessMetrics stringInstance1Dup = cacheMetrics.forInstance(String.class);
CacheAccessMetrics stringInstance2 = cacheMetrics.forInstance(String.class);
CacheAccessMetrics integerInstance1 = cacheMetrics.forInstance(Integer.class);
CacheAccessMetrics integerInstance2 = cacheMetrics.forInstance(Integer.class);
assertThat(stringInstance1).isSameAs(stringInstance1Dup);
assertThat(stringInstance1).isSameAs(stringInstance2);
assertThat(integerInstance1).isSameAs(integerInstance2);
assertThat(stringInstance1).isNotSameAs(integerInstance1);
}
}

6
test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java
View File

@ -77,6 +77,7 @@ import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.statements.TransactionStatement;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.metrics.AccordStateCacheMetrics;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ClientState;
@ -370,7 +371,8 @@ public class AccordTestUtils
holder,
journal,
loadExecutor,
saveExecutor);
saveExecutor,
new AccordStateCacheMetrics(AccordCommandStores.ACCORD_STATE_CACHE + System.currentTimeMillis()));
holder.set(result);
result.updateRangesForEpoch();
return result;
@ -389,7 +391,7 @@ public class AccordTestUtils
Node.Id node = new Id(1);
Topology topology = new Topology(1, new Shard(range, Lists.newArrayList(node), Sets.newHashSet(node), Collections.emptySet()));
AccordCommandStore store = createAccordCommandStore(node, now, topology, loadExecutor, saveExecutor);
store.execute(PreLoadContext.empty(), safeStore -> ((AccordCommandStore)safeStore.commandStore()).setCacheSize(1 << 20));
store.execute(PreLoadContext.empty(), safeStore -> ((AccordCommandStore)safeStore.commandStore()).setCapacity(1 << 20));
return store;
}

4
test/unit/org/apache/cassandra/service/accord/async/AsyncLoaderTest.java
View File

@ -84,7 +84,7 @@ public class AsyncLoaderTest
AccordCommandStore commandStore =
createAccordCommandStore(clock::incrementAndGet, "ks", "tbl", executor, executor);
AccordStateCache.Instance<TxnId, Command, AccordSafeCommand> commandCache = commandStore.commandCache();
commandStore.executeBlocking(() -> commandStore.setCacheSize(1024));
commandStore.executeBlocking(() -> commandStore.setCapacity(1024));
AccordStateCache.Instance<RoutableKey, CommandsForKey, AccordSafeCommandsForKey> cfkCache = commandStore.commandsForKeyCache();
TxnId txnId = txnId(1, clock.incrementAndGet(), 1);
@ -229,7 +229,7 @@ public class AsyncLoaderTest
ManualExecutor executor = new ManualExecutor();
AccordCommandStore commandStore =
createAccordCommandStore(clock::incrementAndGet, "ks", "tbl", executor, executor);
commandStore.executor().submit(() -> commandStore.setCacheSize(1024)).get();
commandStore.executor().submit(() -> commandStore.setCapacity(1024)).get();
AccordStateCache.Instance<TxnId, Command, AccordSafeCommand> commandCache = commandStore.commandCache();
AccordStateCache.Instance<RoutableKey, CommandsForKey, AccordSafeCommandsForKey> cfkCache = commandStore.commandsForKeyCache();
TxnId txnId = txnId(1, clock.incrementAndGet(), 1);

8
test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java
View File

@ -223,9 +223,9 @@ public class AsyncOperationTest
// clear cache
commandStore.executeBlocking(() -> {
long cacheSize = commandStore.getCacheSize();
commandStore.setCacheSize(0);
commandStore.setCacheSize(cacheSize);
long cacheSize = commandStore.capacity();
commandStore.setCapacity(0);
commandStore.setCapacity(cacheSize);
commandStore.cache().awaitSaveResults();
});
@ -312,7 +312,7 @@ public class AsyncOperationTest
// all txn use the same key; 0
Keys keys = keys(Schema.instance.getTableMetadata("ks", "tbl"), 0);
AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl");
commandStore.executeBlocking(() -> commandStore.setCacheSize(0));
commandStore.executeBlocking(() -> commandStore.setCapacity(0));
Gen<TxnId> txnIdGen = rs -> txnId(1, clock.incrementAndGet(), 1);
qt().withPure(false).withExamples(50).forAll(Gens.random(), Gens.lists(txnIdGen).ofSizeBetween(1, 10)).check((rs, ids) -> {