diff --git a/NEWS.txt b/NEWS.txt index 408750ad5e..2509faef8b 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -79,6 +79,8 @@ New features More updates and documentation to follow. - New Guardrails added: - Whether bulk loading of SSTables is allowed. + - nodetool tpstats can display core pool size, max pool size and max tasks queued if --verbose / -v flag is specified. + system_views.thread_pools adds core_pool_size, max_pool_size and max_tasks_queued columns. Upgrading --------- diff --git a/doc/modules/cassandra/pages/managing/operating/virtualtables.adoc b/doc/modules/cassandra/pages/managing/operating/virtualtables.adoc index 807d1ab7f4..f4a5608507 100644 --- a/doc/modules/cassandra/pages/managing/operating/virtualtables.adoc +++ b/doc/modules/cassandra/pages/managing/operating/virtualtables.adoc @@ -379,6 +379,9 @@ Thread pool information includes active tasks, active tasks limit, blocked tasks, blocked tasks all time, completed tasks, and pending tasks. A query on the `thread_pools` returns following details: +From 5.1 (CASSANDRA-19289), this table also displays core pool size, max pool size and max queued tasks values (visible +from `nodetool tpstats --verbose` as well.) + [source, console] ---- cqlsh:system_views> select * from system_views.thread_pools; diff --git a/src/java/org/apache/cassandra/concurrent/ThreadPoolExecutorBase.java b/src/java/org/apache/cassandra/concurrent/ThreadPoolExecutorBase.java index b90485abe6..9975952123 100644 --- a/src/java/org/apache/cassandra/concurrent/ThreadPoolExecutorBase.java +++ b/src/java/org/apache/cassandra/concurrent/ThreadPoolExecutorBase.java @@ -173,6 +173,11 @@ public class ThreadPoolExecutorBase extends ThreadPoolExecutor implements Resiza setMaximumPoolSize(number); } + public int getMaxTasksQueued() + { + return getQueue().remainingCapacity(); + } + @Override public NamedThreadFactory getThreadFactory() { diff --git a/src/java/org/apache/cassandra/concurrent/ThreadPoolExecutorPlus.java b/src/java/org/apache/cassandra/concurrent/ThreadPoolExecutorPlus.java index ad735d945b..97cf0e5b1b 100644 --- a/src/java/org/apache/cassandra/concurrent/ThreadPoolExecutorPlus.java +++ b/src/java/org/apache/cassandra/concurrent/ThreadPoolExecutorPlus.java @@ -116,10 +116,4 @@ public class ThreadPoolExecutorPlus extends ThreadPoolExecutorBase implements Ex { return taskFactory.toSubmit(callable); } - - @Override - public int getMaxTasksQueued() - { - return getQueue().size(); - } } diff --git a/src/java/org/apache/cassandra/db/virtual/ThreadPoolsTable.java b/src/java/org/apache/cassandra/db/virtual/ThreadPoolsTable.java index 778e46f8f4..4bd4624d58 100644 --- a/src/java/org/apache/cassandra/db/virtual/ThreadPoolsTable.java +++ b/src/java/org/apache/cassandra/db/virtual/ThreadPoolsTable.java @@ -36,6 +36,9 @@ final class ThreadPoolsTable extends AbstractVirtualTable private static final String COMPLETED_TASKS = "completed_tasks"; private static final String BLOCKED_TASKS = "blocked_tasks"; private static final String BLOCKED_TASKS_ALL_TIME = "blocked_tasks_all_time"; + private static final String CORE_POOL_SIZE = "core_pool_size"; + private static final String MAX_POOL_SIZE = "max_pool_size"; + private static final String MAX_TASKS_QUEUED = "max_tasks_queued"; ThreadPoolsTable(String keyspace) { @@ -49,6 +52,9 @@ final class ThreadPoolsTable extends AbstractVirtualTable .addRegularColumn(COMPLETED_TASKS, LongType.instance) .addRegularColumn(BLOCKED_TASKS, LongType.instance) .addRegularColumn(BLOCKED_TASKS_ALL_TIME, LongType.instance) + .addRegularColumn(CORE_POOL_SIZE, Int32Type.instance) + .addRegularColumn(MAX_POOL_SIZE, Int32Type.instance) + .addRegularColumn(MAX_TASKS_QUEUED, Int32Type.instance) .build()); } @@ -80,6 +86,9 @@ final class ThreadPoolsTable extends AbstractVirtualTable .column(PENDING_TASKS, metrics.pendingTasks.getValue()) .column(COMPLETED_TASKS, metrics.completedTasks.getValue()) .column(BLOCKED_TASKS, metrics.currentBlocked.getCount()) - .column(BLOCKED_TASKS_ALL_TIME, metrics.totalBlocked.getCount()); + .column(BLOCKED_TASKS_ALL_TIME, metrics.totalBlocked.getCount()) + .column(CORE_POOL_SIZE, metrics.corePoolSize.getValue()) + .column(MAX_POOL_SIZE, metrics.maxPoolSize.getValue()) + .column(MAX_TASKS_QUEUED, metrics.maxTasksQueued.getValue()); } } diff --git a/src/java/org/apache/cassandra/metrics/ThreadPoolMetrics.java b/src/java/org/apache/cassandra/metrics/ThreadPoolMetrics.java index 62ef1775fb..95ab601fe5 100644 --- a/src/java/org/apache/cassandra/metrics/ThreadPoolMetrics.java +++ b/src/java/org/apache/cassandra/metrics/ThreadPoolMetrics.java @@ -38,6 +38,7 @@ public class ThreadPoolMetrics public static final String COMPLETED_TASKS = "CompletedTasks"; public static final String CURRENTLY_BLOCKED_TASKS = "CurrentlyBlockedTasks"; public static final String TOTAL_BLOCKED_TASKS = "TotalBlockedTasks"; + public static final String CORE_POOL_SIZE = "CorePoolSize"; public static final String MAX_POOL_SIZE = "MaxPoolSize"; public static final String MAX_TASKS_QUEUED = "MaxTasksQueued"; @@ -59,6 +60,9 @@ public class ThreadPoolMetrics /** Number of tasks that had blocked before being accepted (or rejected). */ public final Counter totalBlocked; + /** Number of threads always available */ + public final Gauge corePoolSize; + /** Maximum number of threads before it will start queuing tasks */ public final Gauge maxPoolSize; @@ -85,6 +89,7 @@ public class ThreadPoolMetrics activeTasks = executor::getActiveTaskCount; pendingTasks = executor::getPendingTaskCount; completedTasks = executor::getCompletedTaskCount; + corePoolSize = executor::getCorePoolSize; maxPoolSize = executor::getMaximumPoolSize; maxTasksQueued = executor::getMaxTasksQueued; } @@ -96,6 +101,7 @@ public class ThreadPoolMetrics Metrics.register(makeMetricName(path, poolName, COMPLETED_TASKS), completedTasks); Metrics.register(makeMetricName(path, poolName, CURRENTLY_BLOCKED_TASKS), currentBlocked); Metrics.register(makeMetricName(path, poolName, TOTAL_BLOCKED_TASKS), totalBlocked); + Metrics.register(makeMetricName(path, poolName, CORE_POOL_SIZE), corePoolSize); Metrics.register(makeMetricName(path, poolName, MAX_POOL_SIZE), maxPoolSize); Metrics.register(makeMetricName(path, poolName, MAX_TASKS_QUEUED), maxTasksQueued); return Metrics.register(this); @@ -108,6 +114,7 @@ public class ThreadPoolMetrics Metrics.remove(makeMetricName(path, poolName, COMPLETED_TASKS)); Metrics.remove(makeMetricName(path, poolName, CURRENTLY_BLOCKED_TASKS)); Metrics.remove(makeMetricName(path, poolName, TOTAL_BLOCKED_TASKS)); + Metrics.remove(makeMetricName(path, poolName, CORE_POOL_SIZE)); Metrics.remove(makeMetricName(path, poolName, MAX_POOL_SIZE)); Metrics.remove(makeMetricName(path, poolName, MAX_TASKS_QUEUED)); Metrics.remove(this); diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java index 1e5c7e3d9b..7b8fd1e096 100644 --- a/src/java/org/apache/cassandra/tools/NodeProbe.java +++ b/src/java/org/apache/cassandra/tools/NodeProbe.java @@ -1860,7 +1860,9 @@ public class NodeProbe implements AutoCloseable case ThreadPoolMetrics.ACTIVE_TASKS: case ThreadPoolMetrics.PENDING_TASKS: case ThreadPoolMetrics.COMPLETED_TASKS: + case ThreadPoolMetrics.CORE_POOL_SIZE: case ThreadPoolMetrics.MAX_POOL_SIZE: + case ThreadPoolMetrics.MAX_TASKS_QUEUED: return JMX.newMBeanProxy(mbeanServerConn, oName, CassandraMetricsRegistry.JmxGaugeMBean.class).getValue(); case ThreadPoolMetrics.TOTAL_BLOCKED_TASKS: case ThreadPoolMetrics.CURRENTLY_BLOCKED_TASKS: diff --git a/src/java/org/apache/cassandra/tools/nodetool/TpStats.java b/src/java/org/apache/cassandra/tools/nodetool/TpStats.java index 1d16d8dcb9..ce35b52e81 100644 --- a/src/java/org/apache/cassandra/tools/nodetool/TpStats.java +++ b/src/java/org/apache/cassandra/tools/nodetool/TpStats.java @@ -33,6 +33,11 @@ public class TpStats extends NodeToolCmd description = "Output format (json, yaml)") private String outputFormat = ""; + @Option(title = "verbose", + name = {"-v", "--verbose"}, + description = "Display detailed metrics about thread pool's sizes") + private boolean verbose = false; + @Override public void execute(NodeProbe probe) { @@ -43,6 +48,6 @@ public class TpStats extends NodeToolCmd StatsHolder data = new TpStatsHolder(probe); StatsPrinter printer = TpStatsPrinter.from(outputFormat); - printer.print(data, probe.output().out); + printer.print(data, probe.output().out, verbose); } } diff --git a/src/java/org/apache/cassandra/tools/nodetool/stats/StatsPrinter.java b/src/java/org/apache/cassandra/tools/nodetool/stats/StatsPrinter.java index 86d7af7694..690db258cc 100644 --- a/src/java/org/apache/cassandra/tools/nodetool/stats/StatsPrinter.java +++ b/src/java/org/apache/cassandra/tools/nodetool/stats/StatsPrinter.java @@ -33,6 +33,11 @@ import org.yaml.snakeyaml.Yaml; */ public interface StatsPrinter { + default void print(T data, PrintStream out, boolean verbose) + { + print(data, out); + } + void print(T data, PrintStream out); static class JsonPrinter implements StatsPrinter diff --git a/src/java/org/apache/cassandra/tools/nodetool/stats/TpStatsHolder.java b/src/java/org/apache/cassandra/tools/nodetool/stats/TpStatsHolder.java index db9c298865..4a05524664 100644 --- a/src/java/org/apache/cassandra/tools/nodetool/stats/TpStatsHolder.java +++ b/src/java/org/apache/cassandra/tools/nodetool/stats/TpStatsHolder.java @@ -48,6 +48,9 @@ public class TpStatsHolder implements StatsHolder threadPool.put("CompletedTasks", probe.getThreadPoolMetric(tp.getKey(), tp.getValue(), "CompletedTasks")); threadPool.put("CurrentlyBlockedTasks", probe.getThreadPoolMetric(tp.getKey(), tp.getValue(), "CurrentlyBlockedTasks")); threadPool.put("TotalBlockedTasks", probe.getThreadPoolMetric(tp.getKey(), tp.getValue(), "TotalBlockedTasks")); + threadPool.put("CorePoolSize", probe.getThreadPoolMetric(tp.getKey(), tp.getValue(), "CorePoolSize")); + threadPool.put("MaxPoolSize", probe.getThreadPoolMetric(tp.getKey(), tp.getValue(), "MaxPoolSize")); + threadPool.put("MaxTasksQueued", probe.getThreadPoolMetric(tp.getKey(), tp.getValue(), "MaxTasksQueued")); threadPools.put(tp.getValue(), threadPool); } result.put("ThreadPools", threadPools); diff --git a/src/java/org/apache/cassandra/tools/nodetool/stats/TpStatsPrinter.java b/src/java/org/apache/cassandra/tools/nodetool/stats/TpStatsPrinter.java index 252adf000c..fd7dbf64c7 100644 --- a/src/java/org/apache/cassandra/tools/nodetool/stats/TpStatsPrinter.java +++ b/src/java/org/apache/cassandra/tools/nodetool/stats/TpStatsPrinter.java @@ -19,6 +19,7 @@ package org.apache.cassandra.tools.nodetool.stats; import java.io.PrintStream; +import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; import java.util.List; @@ -49,21 +50,35 @@ public class TpStatsPrinter public static class DefaultPrinter implements StatsPrinter { @Override - public void print(TpStatsHolder data, PrintStream out) + public void print(TpStatsHolder data, PrintStream out, boolean verbose) { final TableBuilder poolBuilder = new TableBuilder(); - poolBuilder.add("Pool Name", "Active", "Pending", "Completed", "Blocked", "All time blocked"); + + if (verbose) + poolBuilder.add("Pool Name", "Active", "Pending", "Completed", "Blocked", "All time blocked", "Core Pool Size", "Max Pool Size", "Max Tasks Queued"); + else + poolBuilder.add("Pool Name", "Active", "Pending", "Completed", "Blocked", "All time blocked"); final Multimap threadPools = data.probe.getThreadPools(); for (final Map.Entry tpool : threadPools.entries()) { - poolBuilder.add(tpool.getValue(), - data.probe.getThreadPoolMetric(tpool.getKey(), tpool.getValue(), "ActiveTasks").toString(), - data.probe.getThreadPoolMetric(tpool.getKey(), tpool.getValue(), "PendingTasks").toString(), - data.probe.getThreadPoolMetric(tpool.getKey(), tpool.getValue(), "CompletedTasks").toString(), - data.probe.getThreadPoolMetric(tpool.getKey(), tpool.getValue(), "CurrentlyBlockedTasks").toString(), - data.probe.getThreadPoolMetric(tpool.getKey(), tpool.getValue(), "TotalBlockedTasks").toString()); + List values = new ArrayList<>(); + values.add(tpool.getValue()); + values.add(data.probe.getThreadPoolMetric(tpool.getKey(), tpool.getValue(), "ActiveTasks").toString()); + values.add(data.probe.getThreadPoolMetric(tpool.getKey(), tpool.getValue(), "PendingTasks").toString()); + values.add(data.probe.getThreadPoolMetric(tpool.getKey(), tpool.getValue(), "CompletedTasks").toString()); + values.add(data.probe.getThreadPoolMetric(tpool.getKey(), tpool.getValue(), "CurrentlyBlockedTasks").toString()); + values.add(data.probe.getThreadPoolMetric(tpool.getKey(), tpool.getValue(), "TotalBlockedTasks").toString()); + + if (verbose) + { + values.add(data.probe.getThreadPoolMetric(tpool.getKey(), tpool.getValue(), "CorePoolSize").toString()); + values.add(data.probe.getThreadPoolMetric(tpool.getKey(), tpool.getValue(), "MaxPoolSize").toString()); + values.add(data.probe.getThreadPoolMetric(tpool.getKey(), tpool.getValue(), "MaxTasksQueued").toString()); + } + + poolBuilder.add(values); } poolBuilder.printTo(out); @@ -111,5 +126,11 @@ public class TpStatsPrinter droppedBuilder.printTo(out); } + + @Override + public void print(TpStatsHolder data, PrintStream out) + { + print(data, out, false); + } } } diff --git a/test/unit/org/apache/cassandra/metrics/ThreadPoolMetricsTest.java b/test/unit/org/apache/cassandra/metrics/ThreadPoolMetricsTest.java index 5c80b8ef0d..7ae68468f6 100644 --- a/test/unit/org/apache/cassandra/metrics/ThreadPoolMetricsTest.java +++ b/test/unit/org/apache/cassandra/metrics/ThreadPoolMetricsTest.java @@ -20,15 +20,20 @@ package org.apache.cassandra.metrics; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; import java.util.function.Supplier; import org.junit.Test; import org.apache.cassandra.Util; -import org.apache.cassandra.concurrent.*; +import org.apache.cassandra.concurrent.ExecutorPlus; +import org.apache.cassandra.concurrent.SEPExecutor; +import org.apache.cassandra.concurrent.SharedExecutorPool; +import org.apache.cassandra.concurrent.ThreadPoolExecutorJMXAdapter; +import org.apache.cassandra.concurrent.ThreadPoolExecutorPlus; import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory; -import static org.junit.Assert.*; +import static org.junit.Assert.assertEquals; public class ThreadPoolMetricsTest { @@ -40,7 +45,7 @@ public class ThreadPoolMetricsTest .configurePooled("ThreadPoolMetricsTest-1", 2) .withQueueLimit(2) .build(); - testMetricsWithNoBlockedThreads(executor, ((ThreadPoolExecutorJMXAdapter)executor.onShutdown()).metrics()); + testMetricsWithNoBlockedThreads(executor, ((ThreadPoolExecutorJMXAdapter)executor.onShutdown()).metrics(), 2); } @Test @@ -61,12 +66,13 @@ public class ThreadPoolMetricsTest "ThreadPoolMetricsTest-3", "internal"); - testMetricsWithNoBlockedThreads(executor, executor.metrics); + testMetricsWithNoBlockedThreads(executor, executor.metrics, Integer.MAX_VALUE); } private static void testMetricsWithBlockedThreads(ExecutorPlus threadPool, ThreadPoolMetrics metrics) { assertEquals(2, metrics.maxPoolSize.getValue().intValue()); + assertEquals(2, metrics.maxTasksQueued.getValue().intValue()); spinAssertEquals(0, metrics.activeTasks::getValue); spinAssertEquals(0L, metrics.completedTasks::getValue); @@ -88,6 +94,7 @@ public class ThreadPoolMetricsTest spinAssertEquals(0, metrics.pendingTasks::getValue); spinAssertEquals(0L, metrics.currentBlocked::getCount); spinAssertEquals(0L, metrics.totalBlocked::getCount); + spinAssertEquals(2, metrics.maxTasksQueued::getValue); // There are no threads available any more the 2 next tasks should go into the queue threadPool.execute(task3); @@ -98,6 +105,7 @@ public class ThreadPoolMetricsTest spinAssertEquals(2, metrics.pendingTasks::getValue); spinAssertEquals(0L, metrics.currentBlocked::getCount); spinAssertEquals(0L, metrics.totalBlocked::getCount); + spinAssertEquals(0, metrics.maxTasksQueued::getValue); // The queue is full the 2 next task should go into blocked and block the thread BlockingTask task5 = new BlockingTask(); @@ -117,6 +125,7 @@ public class ThreadPoolMetricsTest spinAssertEquals(2, metrics.pendingTasks::getValue); spinAssertEquals(1L, metrics.currentBlocked::getCount); spinAssertEquals(1L, metrics.totalBlocked::getCount); + spinAssertEquals(0, metrics.maxTasksQueued::getValue); new Thread(() -> { @@ -131,6 +140,7 @@ public class ThreadPoolMetricsTest spinAssertEquals(2, metrics.pendingTasks::getValue); spinAssertEquals(2L, metrics.currentBlocked::getCount); spinAssertEquals(2L, metrics.totalBlocked::getCount); + spinAssertEquals(0, metrics.maxTasksQueued::getValue); // Allowing first task to complete task1.allowToComplete(); @@ -142,6 +152,7 @@ public class ThreadPoolMetricsTest spinAssertEquals(1L, metrics.currentBlocked::getCount); spinAssertEquals(2L, metrics.totalBlocked::getCount); spinAssertEquals(1, blockedThreads::get); + spinAssertEquals(0, metrics.maxTasksQueued::getValue); // Allowing second task to complete task2.allowToComplete(); @@ -153,6 +164,7 @@ public class ThreadPoolMetricsTest spinAssertEquals(0L, metrics.currentBlocked::getCount); spinAssertEquals(2L, metrics.totalBlocked::getCount); spinAssertEquals(0, blockedThreads::get); + spinAssertEquals(0, metrics.maxTasksQueued::getValue); // Allowing third task to complete task3.allowToComplete(); @@ -163,6 +175,7 @@ public class ThreadPoolMetricsTest spinAssertEquals(1, metrics.pendingTasks::getValue); spinAssertEquals(0L, metrics.currentBlocked::getCount); spinAssertEquals(2L, metrics.totalBlocked::getCount); + spinAssertEquals(1, metrics.maxTasksQueued::getValue); // Allowing fourth task to complete task4.allowToComplete(); @@ -173,6 +186,7 @@ public class ThreadPoolMetricsTest spinAssertEquals(0, metrics.pendingTasks::getValue); spinAssertEquals(0L, metrics.currentBlocked::getCount); spinAssertEquals(2L, metrics.totalBlocked::getCount); + spinAssertEquals(2, metrics.maxTasksQueued::getValue); // Allowing last tasks to complete task5.allowToComplete(); @@ -183,15 +197,19 @@ public class ThreadPoolMetricsTest spinAssertEquals(0, metrics.pendingTasks::getValue); spinAssertEquals(0L, metrics.currentBlocked::getCount); spinAssertEquals(2L, metrics.totalBlocked::getCount); + spinAssertEquals(2, metrics.maxTasksQueued::getValue); } - private static void testMetricsWithNoBlockedThreads(ExecutorPlus threadPool, ThreadPoolMetrics metrics) + private static void testMetricsWithNoBlockedThreads(ExecutorPlus threadPool, ThreadPoolMetrics metrics, Integer initialMaxQueueSize) { + Integer maxQueueSize = initialMaxQueueSize; + spinAssertEquals(0, metrics.activeTasks::getValue); spinAssertEquals(0L, metrics.completedTasks::getValue); spinAssertEquals(0, metrics.pendingTasks::getValue); spinAssertEquals(0L, metrics.currentBlocked::getCount); spinAssertEquals(0L, metrics.totalBlocked::getCount); + spinAssertEquals(maxQueueSize, metrics.maxTasksQueued::getValue); BlockingTask task1 = new BlockingTask(); BlockingTask task2 = new BlockingTask(); @@ -206,6 +224,7 @@ public class ThreadPoolMetricsTest spinAssertEquals(0, metrics.pendingTasks::getValue); spinAssertEquals(0L, metrics.currentBlocked::getCount); spinAssertEquals(0L, metrics.totalBlocked::getCount); + spinAssertEquals(maxQueueSize, metrics.maxTasksQueued::getValue); threadPool.execute(task2); @@ -214,6 +233,7 @@ public class ThreadPoolMetricsTest spinAssertEquals(0, metrics.pendingTasks::getValue); spinAssertEquals(0L, metrics.currentBlocked::getCount); spinAssertEquals(0L, metrics.totalBlocked::getCount); + spinAssertEquals(maxQueueSize, metrics.maxTasksQueued::getValue); // There are no threads available any more the 2 next tasks should go into the queue threadPool.execute(task3); @@ -223,6 +243,8 @@ public class ThreadPoolMetricsTest spinAssertEquals(1, metrics.pendingTasks::getValue); spinAssertEquals(0L, metrics.currentBlocked::getCount); spinAssertEquals(0L, metrics.totalBlocked::getCount); + maxQueueSize = resolveMaxTasksQueued(maxQueueSize, v -> --v); + spinAssertEquals(maxQueueSize, metrics.maxTasksQueued::getValue); threadPool.execute(task4); @@ -231,6 +253,8 @@ public class ThreadPoolMetricsTest spinAssertEquals(2, metrics.pendingTasks::getValue); spinAssertEquals(0L, metrics.currentBlocked::getCount); spinAssertEquals(0L, metrics.totalBlocked::getCount); + maxQueueSize = resolveMaxTasksQueued(maxQueueSize, v -> --v); + spinAssertEquals(maxQueueSize, metrics.maxTasksQueued::getValue); // Allowing first task to complete task1.allowToComplete(); @@ -241,6 +265,8 @@ public class ThreadPoolMetricsTest spinAssertEquals(1, metrics.pendingTasks::getValue); spinAssertEquals(0L, metrics.currentBlocked::getCount); spinAssertEquals(0L, metrics.totalBlocked::getCount); + maxQueueSize = resolveMaxTasksQueued(maxQueueSize, v -> ++v); + spinAssertEquals(maxQueueSize, metrics.maxTasksQueued::getValue); // Allowing second task to complete task2.allowToComplete(); @@ -251,6 +277,8 @@ public class ThreadPoolMetricsTest spinAssertEquals(0, metrics.pendingTasks::getValue); spinAssertEquals(0L, metrics.currentBlocked::getCount); spinAssertEquals(0L, metrics.totalBlocked::getCount); + maxQueueSize = resolveMaxTasksQueued(maxQueueSize, v -> ++v); + spinAssertEquals(maxQueueSize, metrics.maxTasksQueued::getValue); // Allowing third task to complete task3.allowToComplete(); @@ -260,6 +288,7 @@ public class ThreadPoolMetricsTest spinAssertEquals(0, metrics.pendingTasks::getValue); spinAssertEquals(0L, metrics.currentBlocked::getCount); spinAssertEquals(0L, metrics.totalBlocked::getCount); + spinAssertEquals(maxQueueSize, metrics.maxTasksQueued::getValue); // Allowing fourth task to complete task4.allowToComplete(); @@ -269,6 +298,21 @@ public class ThreadPoolMetricsTest spinAssertEquals(0, metrics.pendingTasks::getValue); spinAssertEquals(0L, metrics.currentBlocked::getCount); spinAssertEquals(0L, metrics.totalBlocked::getCount); + spinAssertEquals(maxQueueSize, metrics.maxTasksQueued::getValue); + } + + private static Integer resolveMaxTasksQueued(Integer value, Function operation) + { + if (value == Integer.MAX_VALUE) + { + // SEPExecutor doesn't care + return Integer.MAX_VALUE; + } + else + { + value = operation.apply(value); + return value; + } } private static void spinAssertEquals(Object expected, Supplier actualSupplier) diff --git a/test/unit/org/apache/cassandra/tools/nodetool/TpStatsTest.java b/test/unit/org/apache/cassandra/tools/nodetool/TpStatsTest.java index e9548fbdbf..7fb626f409 100644 --- a/test/unit/org/apache/cassandra/tools/nodetool/TpStatsTest.java +++ b/test/unit/org/apache/cassandra/tools/nodetool/TpStatsTest.java @@ -67,7 +67,7 @@ public class TpStatsTest extends CQLTester " [(-pp | --print-port)] [(-pw | --password )]\n" + " [(-pwf | --password-file )]\n" + " [(-u | --username )] tpstats\n" + - " [(-F | --format )]\n" + + " [(-F | --format )] [(-v | --verbose)]\n" + "\n" + "OPTIONS\n" + " -F , --format \n" + @@ -89,7 +89,10 @@ public class TpStatsTest extends CQLTester " Path to the JMX password file\n" + "\n" + " -u , --username \n" + - " Remote jmx agent username\n" + + " Remote jmx agent username\n" + + "\n" + + " -v, --verbose\n" + + " Display detailed metrics about thread pool's sizes\n" + "\n" + "\n"; assertThat(tool.getStdout()).isEqualTo(help);