This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
The following commit(s) were added to refs/heads/main by this push:
new c6b22a19c2 ARTEMIS-5557 support Micrometer ExecutorService metrics
c6b22a19c2 is described below
commit c6b22a19c2a340546c267cfac896fd6bbf3307a2
Author: Justin Bertram <[email protected]>
AuthorDate: Fri Jun 27 23:50:43 2025 -0500
ARTEMIS-5557 support Micrometer ExecutorService metrics
Aside from supporting Micrometer ExecutorService metrics this commit
improves the consistency and clarity of both the documentation and code
related to thread pooling. Changes include:
- Also support Netty EventExecutor metrics
- Remove redundant "threads" from thread group names
- Use kebab case in ActiveMQThreadFactory for consistent naming
- Use "activemq-" prefix in ActiveMQThreadFactory for consistent
naming
- Use consistent broker identification in thread group names
- Remove unused executor from SequentialFile classes
- Big refactor of thread pooling documentation
- Update release notes with relevant naming changes
- Deduplicate some repeated code
- Pass custom ActiveMQThreadFactory and Scheduler to Jetty for
consistent thread naming
- Name Netty threads according to corresponding acceptor
---
.../apache/activemq/artemis/cli/commands/Run.java | 2 +-
.../core/server/ActiveMQScheduledComponent.java | 4 +-
.../artemis/core/server/NetworkHealthCheck.java | 2 +-
.../artemis/utils/ActiveMQThreadFactory.java | 2 +-
.../apache/activemq/artemis/utils/StringUtil.java | 19 ++++
.../utils/critical/CriticalAnalyzerImpl.java | 2 +-
.../activemq/artemis/utils}/StringUtilTest.java | 8 +-
.../api/config/ActiveMQDefaultConfiguration.java | 7 ++
.../artemis/api/core/client/ActiveMQClient.java | 6 +-
.../core/client/impl/ServerLocatorImpl.java | 6 +-
.../artemis/core/cluster/DiscoveryGroup.java | 7 +-
.../remoting/impl/netty/SharedEventLoopGroup.java | 2 +-
.../artemis/core/io/AbstractSequentialFile.java | 8 +-
.../core/io/AbstractSequentialFileFactory.java | 35 ------
.../artemis/core/io/aio/AIOSequentialFile.java | 12 +--
.../core/io/aio/AIOSequentialFileFactory.java | 4 +-
.../artemis/core/io/nio/NIOSequentialFile.java | 10 +-
.../core/io/nio/NIOSequentialFileFactory.java | 2 +-
.../artemis/core/journal/impl/JournalImpl.java | 2 +-
.../artemis/ra/inflow/ActiveMQActivation.java | 2 +-
.../artemis/core/config/MetricsConfiguration.java | 10 ++
.../deployers/impl/FileConfigurationParser.java | 2 +
.../remoting/impl/invm/InVMAcceptorFactory.java | 5 +-
.../core/remoting/impl/netty/NettyAcceptor.java | 52 +++++----
.../remoting/impl/netty/NettyAcceptorFactory.java | 7 +-
.../remoting/server/impl/RemotingServiceImpl.java | 34 +++---
.../artemis/core/server/ActiveMQServer.java | 2 +
.../core/server/impl/ActiveMQServerImpl.java | 71 +++++++------
.../server/impl/ReplicationPrimaryActivation.java | 10 +-
.../impl/SharedNothingPrimaryActivation.java | 10 +-
.../core/server/metrics/BrokerMetricNames.java | 4 +
.../core/server/metrics/MetricsManager.java | 77 ++++++++++++--
.../metrics/NettyPooledAllocatorMetrics.java | 4 +-
.../artemis/spi/core/remoting/Acceptor.java | 6 ++
.../artemis/spi/core/remoting/AcceptorFactory.java | 5 +-
.../resources/schema/artemis-configuration.xsd | 8 ++
.../config/impl/DefaultsFileConfigurationTest.java | 2 +
.../core/config/impl/FileConfigurationTest.java | 1 +
.../activemq/artemis/uri/AcceptorParserTest.java | 2 +-
.../resources/ConfigurationTest-full-config.xml | 1 +
.../ConfigurationTest-xinclude-config.xml | 1 +
...gurationTest-xinclude-schema-config-metrics.xml | 1 +
artemis-server/src/test/resources/metrics.xml | 1 +
.../tests/extensions/ThreadLeakCheckDelegate.java | 2 +-
.../artemis/component/WebServerComponent.java | 11 +-
docs/user-manual/configuration-index.adoc | 2 +-
docs/user-manual/metrics.adoc | 79 ++++++++++----
docs/user-manual/thread-pooling.adoc | 117 +++++++++++++++------
docs/user-manual/versions.adoc | 53 ++++++++++
.../largemessage/ServerLargeMessageTest.java | 2 +-
.../plugin/ExecutorServiceMetricsTest.java | 90 ++++++++++++++++
.../SharedNothingReplicationFlowControlTest.java | 20 ++--
.../tests/smoke/nettynative/NettyNativeTest.java | 7 +-
.../unit/core/journal/impl/TimedBufferTest.java | 2 +-
.../impl/netty/NettyAcceptorFactoryTest.java | 2 +-
.../remoting/impl/netty/NettyAcceptorTest.java | 8 +-
.../server/impl/fake/FakeAcceptorFactory.java | 5 +-
57 files changed, 610 insertions(+), 248 deletions(-)
diff --git
a/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/Run.java
b/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/Run.java
index 7eb70011b1..c89c2ad412 100644
---
a/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/Run.java
+++
b/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/Run.java
@@ -179,7 +179,7 @@ public class Run extends LockAbstract {
}
}
- shutdownTimer = new Timer("ActiveMQ Artemis Server Shutdown Timer",
true);
+ shutdownTimer = new Timer("activemq-shutdown-timer", true);
shutdownTimer.scheduleAtFixedRate(new TimerTask() {
@Override
public void run() {
diff --git
a/artemis-commons/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQScheduledComponent.java
b/artemis-commons/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQScheduledComponent.java
index 31089a7ad0..2a97640099 100644
---
a/artemis-commons/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQScheduledComponent.java
+++
b/artemis-commons/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQScheduledComponent.java
@@ -16,6 +16,7 @@
*/
package org.apache.activemq.artemis.core.server;
+import java.lang.invoke.MethodHandles;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.Objects;
@@ -30,7 +31,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.activemq.artemis.utils.ActiveMQThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.lang.invoke.MethodHandles;
/**
* This is for components with a scheduled at a fixed rate.
@@ -184,7 +184,7 @@ public abstract class ActiveMQScheduledComponent implements
ActiveMQComponent, R
}
protected ActiveMQThreadFactory getThreadFactory() {
- return new ActiveMQThreadFactory(this.getClass().getSimpleName() +
"-scheduled-threads", false, getThisClassLoader());
+ return new ActiveMQThreadFactory(this.getClass().getSimpleName() +
"-scheduled", false, getThisClassLoader());
}
private ClassLoader getThisClassLoader() {
diff --git
a/artemis-commons/src/main/java/org/apache/activemq/artemis/core/server/NetworkHealthCheck.java
b/artemis-commons/src/main/java/org/apache/activemq/artemis/core/server/NetworkHealthCheck.java
index 4d29a5e4a8..0272ef8f37 100644
---
a/artemis-commons/src/main/java/org/apache/activemq/artemis/core/server/NetworkHealthCheck.java
+++
b/artemis-commons/src/main/java/org/apache/activemq/artemis/core/server/NetworkHealthCheck.java
@@ -160,7 +160,7 @@ public class NetworkHealthCheck extends
ActiveMQScheduledComponent {
@Override
protected ActiveMQThreadFactory getThreadFactory() {
- return new ActiveMQThreadFactory("NetworkChecker", "Network-Checker-",
false, getThisClassLoader());
+ return new ActiveMQThreadFactory("network-checker", false,
getThisClassLoader());
}
diff --git
a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/ActiveMQThreadFactory.java
b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/ActiveMQThreadFactory.java
index 6edeaf5348..d829e593a9 100644
---
a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/ActiveMQThreadFactory.java
+++
b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/ActiveMQThreadFactory.java
@@ -62,7 +62,7 @@ public final class ActiveMQThreadFactory implements
ThreadFactory {
* @param tccl the context class loader of newly created threads
*/
public ActiveMQThreadFactory(final String groupName, String prefix, final
boolean daemon, final ClassLoader tccl) {
- this.groupName = groupName;
+ this.groupName = "activemq-" +
StringUtil.convertPascalCaseToKebabCase(groupName);
this.prefix = prefix;
diff --git
a/artemis-core-client/src/main/java/org/apache/activemq/artemis/utils/StringUtil.java
b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/StringUtil.java
similarity index 79%
rename from
artemis-core-client/src/main/java/org/apache/activemq/artemis/utils/StringUtil.java
rename to
artemis-commons/src/main/java/org/apache/activemq/artemis/utils/StringUtil.java
index 5ec17a897f..ae1b183b16 100644
---
a/artemis-core-client/src/main/java/org/apache/activemq/artemis/utils/StringUtil.java
+++
b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/StringUtil.java
@@ -59,4 +59,23 @@ public class StringUtil {
}
return list;
}
+
+ public static String convertPascalCaseToKebabCase(String input) {
+ if (input == null || input.isEmpty()) {
+ return input;
+ }
+ StringBuilder builder = new StringBuilder();
+ for (int i = 0; i < input.length(); i++) {
+ char c = input.charAt(i);
+ if (Character.isUpperCase(c)) {
+ if (i > 0) {
+ builder.append('-');
+ }
+ builder.append(Character.toLowerCase(c));
+ } else {
+ builder.append(c);
+ }
+ }
+ return builder.toString();
+ }
}
diff --git
a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalAnalyzerImpl.java
b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalAnalyzerImpl.java
index 0e41034de9..589ac2bcfa 100644
---
a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalAnalyzerImpl.java
+++
b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalAnalyzerImpl.java
@@ -59,7 +59,7 @@ public class CriticalAnalyzerImpl implements CriticalAnalyzer
{
@Override
protected ActiveMQThreadFactory getThreadFactory() {
- return new ActiveMQThreadFactory("CriticalAnalyzer",
"Critical-Analyzer-", true, getThisClassLoader());
+ return new ActiveMQThreadFactory("critical-analyzer", true,
getThisClassLoader());
}
private ClassLoader getThisClassLoader() {
diff --git
a/artemis-core-client/src/test/java/org/apache/activemq/artemis/util/StringUtilTest.java
b/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/StringUtilTest.java
similarity index 92%
rename from
artemis-core-client/src/test/java/org/apache/activemq/artemis/util/StringUtilTest.java
rename to
artemis-commons/src/test/java/org/apache/activemq/artemis/utils/StringUtilTest.java
index c73b455fc0..592accd0f6 100644
---
a/artemis-core-client/src/test/java/org/apache/activemq/artemis/util/StringUtilTest.java
+++
b/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/StringUtilTest.java
@@ -14,14 +14,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.activemq.artemis.util;
+package org.apache.activemq.artemis.utils;
import static org.junit.jupiter.api.Assertions.assertEquals;
import java.util.ArrayList;
import java.util.List;
-import org.apache.activemq.artemis.utils.StringUtil;
import org.junit.jupiter.api.Test;
public class StringUtilTest {
@@ -65,4 +64,9 @@ public class StringUtilTest {
assertEquals("yellow", result.get(2));
assertEquals("green", result.get(3));
}
+
+ @Test
+ public void testPascalToKebab() throws Exception {
+ assertEquals("pascal-to-kebab",
StringUtil.convertPascalCaseToKebabCase("PascalToKebab"));
+ }
}
diff --git
a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/config/ActiveMQDefaultConfiguration.java
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/config/ActiveMQDefaultConfiguration.java
index 4f71d57233..c7f7af4ec5 100644
---
a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/config/ActiveMQDefaultConfiguration.java
+++
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/config/ActiveMQDefaultConfiguration.java
@@ -652,6 +652,9 @@ public final class ActiveMQDefaultConfiguration {
// Whether or not to report JVM thread metrics
private static final boolean DEFAULT_JVM_THREAD_METRICS = false;
+ // Whether or not to report executor service metrics
+ private static final boolean DEFAULT_EXECUTOR_SERVICE_METRICS = false;
+
public static final String DEFAULT_UUID_NAMESPACE = "";
@Deprecated(forRemoval = true)
@@ -1879,6 +1882,10 @@ public final class ActiveMQDefaultConfiguration {
return DEFAULT_JVM_THREAD_METRICS;
}
+ public static boolean getDefaultExecutorServiceMetrics() {
+ return DEFAULT_EXECUTOR_SERVICE_METRICS;
+ }
+
public static String getDefaultUuidNamespace() {
return DEFAULT_UUID_NAMESPACE;
}
diff --git
a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/client/ActiveMQClient.java
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/client/ActiveMQClient.java
index 2954942cc4..5c179ae7fa 100644
---
a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/client/ActiveMQClient.java
+++
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/client/ActiveMQClient.java
@@ -245,12 +245,12 @@ public final class ActiveMQClient {
}
public static synchronized ExecutorService getGlobalThreadPool() {
- globalThreadPool = internalGetGlobalThreadPool(globalThreadPool,
"ActiveMQ-client-global-threads", ActiveMQClient.globalThreadPoolSize);
+ globalThreadPool = internalGetGlobalThreadPool(globalThreadPool,
"client-global", ActiveMQClient.globalThreadPoolSize);
return globalThreadPool;
}
public static synchronized ExecutorService getGlobalFlowControlThreadPool()
{
- globalFlowControlThreadPool =
internalGetGlobalThreadPool(globalFlowControlThreadPool,
"ActiveMQ-client-global-flow-control-threads",
ActiveMQClient.globalFlowControlThreadPoolSize);
+ globalFlowControlThreadPool =
internalGetGlobalThreadPool(globalFlowControlThreadPool,
"client-global-flow-control", ActiveMQClient.globalFlowControlThreadPoolSize);
return globalFlowControlThreadPool;
}
@@ -269,7 +269,7 @@ public final class ActiveMQClient {
public static synchronized ScheduledExecutorService
getGlobalScheduledThreadPool() {
if (globalScheduledThreadPool == null) {
- ThreadFactory factory =
AccessController.doPrivileged((PrivilegedAction<ThreadFactory>) () -> new
ActiveMQThreadFactory("ActiveMQ-client-global-scheduled-threads", true,
ClientSessionFactoryImpl.class.getClassLoader()));
+ ThreadFactory factory =
AccessController.doPrivileged((PrivilegedAction<ThreadFactory>) () -> new
ActiveMQThreadFactory("client-global-scheduled", true,
ClientSessionFactoryImpl.class.getClassLoader()));
globalScheduledThreadPool = new
ScheduledThreadPoolExecutor(ActiveMQClient.globalScheduledThreadPoolSize,
factory);
}
diff --git
a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ServerLocatorImpl.java
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ServerLocatorImpl.java
index e6d5958fbd..5656711d3c 100644
---
a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ServerLocatorImpl.java
+++
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ServerLocatorImpl.java
@@ -207,21 +207,21 @@ public final class ServerLocatorImpl implements
ServerLocatorInternal, Discovery
} else {
this.shutdownPool = true;
- ThreadFactory factory =
getThreadFactory("ActiveMQ-client-factory-threads-");
+ ThreadFactory factory = getThreadFactory("client-factory-");
if (config.threadPoolMaxSize == -1) {
threadPool = Executors.newCachedThreadPool(factory);
} else {
threadPool = new ActiveMQThreadPoolExecutor(0,
config.threadPoolMaxSize, 60L, TimeUnit.SECONDS, factory);
}
- factory =
getThreadFactory("ActiveMQ-client-factory-flow-control-threads-");
+ factory = getThreadFactory("client-factory-flow-control-");
if (config.flowControlThreadPoolMaxSize == -1) {
flowControlThreadPool = Executors.newCachedThreadPool(factory);
} else {
flowControlThreadPool = new ActiveMQThreadPoolExecutor(0,
config.flowControlThreadPoolMaxSize, 60L, TimeUnit.SECONDS, factory);
}
- factory = getThreadFactory("ActiveMQ-client-factory-pinger-threads-");
+ factory = getThreadFactory("client-factory-pinger-");
scheduledThreadPool =
Executors.newScheduledThreadPool(config.scheduledThreadPoolMaxSize, factory);
}
this.updateArrayActor = new Actor<>(threadPool,
this::internalUpdateArray);
diff --git
a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/cluster/DiscoveryGroup.java
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/cluster/DiscoveryGroup.java
index 3782404c52..fbe37d2a7b 100644
---
a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/cluster/DiscoveryGroup.java
+++
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/cluster/DiscoveryGroup.java
@@ -111,12 +111,7 @@ public final class DiscoveryGroup implements
ActiveMQComponent {
started = true;
- ThreadFactory tfactory = AccessController.doPrivileged(new
PrivilegedAction<>() {
- @Override
- public ThreadFactory run() {
- return new ActiveMQThreadFactory("DiscoveryGroup-" +
System.identityHashCode(this), "activemq-discovery-group-thread-" + name, true,
DiscoveryGroup.class.getClassLoader());
- }
- });
+ ThreadFactory tfactory =
AccessController.doPrivileged((PrivilegedAction<ThreadFactory>) () -> new
ActiveMQThreadFactory("discovery-group-" + name, true,
DiscoveryGroup.class.getClassLoader()));
thread = tfactory.newThread(new DiscoveryRunnable());
diff --git
a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/SharedEventLoopGroup.java
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/SharedEventLoopGroup.java
index e42a24d8d5..4ba51692af 100644
---
a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/SharedEventLoopGroup.java
+++
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/SharedEventLoopGroup.java
@@ -61,7 +61,7 @@ public class SharedEventLoopGroup extends
DelegatingEventLoopGroup {
f.cancel(false);
}
} else {
- instance = new
SharedEventLoopGroup(eventLoopGroupSupplier.apply((ThreadFactory)
AccessController.doPrivileged((PrivilegedAction) () -> new
ActiveMQThreadFactory("ActiveMQ-client-netty-threads", true,
ClientSessionFactoryImpl.class.getClassLoader()))));
+ instance = new
SharedEventLoopGroup(eventLoopGroupSupplier.apply((ThreadFactory)
AccessController.doPrivileged((PrivilegedAction) () -> new
ActiveMQThreadFactory("client-remoting", true,
ClientSessionFactoryImpl.class.getClassLoader()))));
}
instance.channelFactoryCount.incrementAndGet();
return instance;
diff --git
a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/AbstractSequentialFile.java
b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/AbstractSequentialFile.java
index b94198363e..b05694910e 100644
---
a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/AbstractSequentialFile.java
+++
b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/AbstractSequentialFile.java
@@ -18,11 +18,11 @@ package org.apache.activemq.artemis.core.io;
import java.io.File;
import java.io.IOException;
+import java.lang.invoke.MethodHandles;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.file.Files;
import java.util.List;
-import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicLong;
import io.netty.buffer.ByteBuf;
@@ -39,9 +39,8 @@ import
org.apache.activemq.artemis.core.journal.impl.SimpleWaitIOCallback;
import org.apache.activemq.artemis.journal.ActiveMQJournalBundle;
import org.apache.activemq.artemis.journal.ActiveMQJournalLogger;
import org.apache.activemq.artemis.utils.ByteUtil;
-import org.slf4j.LoggerFactory;
-import java.lang.invoke.MethodHandles;
import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public abstract class AbstractSequentialFile implements SequentialFile {
@@ -67,8 +66,7 @@ public abstract class AbstractSequentialFile implements
SequentialFile {
public AbstractSequentialFile(final File directory,
final String file,
- final SequentialFileFactory factory,
- final Executor writerExecutor) {
+ final SequentialFileFactory factory) {
super();
this.file = new File(directory, file);
this.directory = directory;
diff --git
a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/AbstractSequentialFileFactory.java
b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/AbstractSequentialFileFactory.java
index a360c767be..cb4397bda0 100644
---
a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/AbstractSequentialFileFactory.java
+++
b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/AbstractSequentialFileFactory.java
@@ -20,19 +20,12 @@ import java.io.File;
import java.io.FilenameFilter;
import java.io.IOException;
import java.nio.ByteBuffer;
-import java.security.AccessController;
-import java.security.PrivilegedAction;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-import org.apache.activemq.artemis.api.core.ActiveMQInterruptedException;
import org.apache.activemq.artemis.core.io.buffer.TimedBuffer;
import org.apache.activemq.artemis.journal.ActiveMQJournalLogger;
-import org.apache.activemq.artemis.utils.ActiveMQThreadFactory;
import org.apache.activemq.artemis.utils.critical.CriticalAnalyzer;
import org.apache.activemq.artemis.utils.critical.EmptyCriticalAnalyzer;
@@ -62,13 +55,6 @@ public abstract class AbstractSequentialFileFactory
implements SequentialFileFac
protected final CriticalAnalyzer criticalAnalyzer;
- /**
- * Asynchronous writes need to be done at another executor. This needs to
be done at NIO, or else we would have the
- * callers thread blocking for the return. At AIO this is necessary as
context switches on writes would fire flushes
- * at the kernel.
- */
- protected ExecutorService writeExecutor;
-
protected AbstractSequentialFileFactory(final File journalDir,
final boolean buffered,
final int bufferSize,
@@ -149,18 +135,6 @@ public abstract class AbstractSequentialFileFactory
implements SequentialFileFac
if (timedBuffer != null) {
timedBuffer.stop();
}
-
- if (isSupportsCallbacks() && writeExecutor != null) {
- writeExecutor.shutdown();
-
- try {
- if
(!writeExecutor.awaitTermination(AbstractSequentialFileFactory.EXECUTOR_TIMEOUT,
TimeUnit.SECONDS)) {
- ActiveMQJournalLogger.LOGGER.timeoutOnWriterShutdown(new
Exception("trace"));
- }
- } catch (InterruptedException e) {
- throw new ActiveMQInterruptedException(e);
- }
- }
}
@Override
@@ -173,15 +147,6 @@ public abstract class AbstractSequentialFileFactory
implements SequentialFileFac
if (timedBuffer != null) {
timedBuffer.start();
}
-
- if (isSupportsCallbacks()) {
- writeExecutor =
Executors.newSingleThreadExecutor(AccessController.doPrivileged(new
PrivilegedAction<ActiveMQThreadFactory>() {
- @Override
- public ActiveMQThreadFactory run() {
- return new
ActiveMQThreadFactory("ActiveMQ-Asynchronous-Persistent-Writes" +
System.identityHashCode(this), true,
AbstractSequentialFileFactory.class.getClassLoader());
- }
- }));
- }
}
@Override
diff --git
a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/aio/AIOSequentialFile.java
b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/aio/AIOSequentialFile.java
index 0c3012d135..97065efe47 100644
---
a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/aio/AIOSequentialFile.java
+++
b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/aio/AIOSequentialFile.java
@@ -18,9 +18,9 @@ package org.apache.activemq.artemis.core.io.aio;
import java.io.File;
import java.io.IOException;
+import java.lang.invoke.MethodHandles;
import java.nio.ByteBuffer;
import java.util.PriorityQueue;
-import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.activemq.artemis.api.core.ActiveMQException;
@@ -31,9 +31,8 @@ import org.apache.activemq.artemis.core.io.IOCallback;
import org.apache.activemq.artemis.core.io.SequentialFile;
import org.apache.activemq.artemis.core.journal.impl.SimpleWaitIOCallback;
import org.apache.activemq.artemis.nativo.jlibaio.LibaioFile;
-import org.slf4j.LoggerFactory;
-import java.lang.invoke.MethodHandles;
import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* This class is implementing Runnable to reuse a callback to close it.
@@ -69,9 +68,8 @@ public class AIOSequentialFile extends AbstractSequentialFile
{
final int bufferSize,
final long bufferTimeoutMilliseconds,
final File directory,
- final String fileName,
- final Executor writerExecutor) {
- super(directory, fileName, factory, writerExecutor);
+ final String fileName) {
+ super(directory, fileName, factory);
this.aioFactory = factory;
}
@@ -92,7 +90,7 @@ public class AIOSequentialFile extends AbstractSequentialFile
{
@Override
public SequentialFile cloneFile() {
- return new AIOSequentialFile(aioFactory, -1, -1,
getFile().getParentFile(), getFile().getName(), null);
+ return new AIOSequentialFile(aioFactory, -1, -1,
getFile().getParentFile(), getFile().getName());
}
diff --git
a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/aio/AIOSequentialFileFactory.java
b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/aio/AIOSequentialFileFactory.java
index feae694a1f..6917a0013f 100644
---
a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/aio/AIOSequentialFileFactory.java
+++
b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/aio/AIOSequentialFileFactory.java
@@ -138,7 +138,7 @@ public final class AIOSequentialFileFactory extends
AbstractSequentialFileFactor
@Override
public SequentialFile createSequentialFile(final String fileName) {
- return new AIOSequentialFile(this, bufferSize, bufferTimeout,
journalDir, fileName, writeExecutor);
+ return new AIOSequentialFile(this, bufferSize, bufferTimeout,
journalDir, fileName);
}
@Override
@@ -439,7 +439,7 @@ public final class AIOSequentialFileFactory extends
AbstractSequentialFileFactor
private class PollerThread extends Thread {
private PollerThread() {
- super("Apache ActiveMQ Artemis libaio poller");
+ super("activemq-libaio-poller");
}
@Override
diff --git
a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFile.java
b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFile.java
index 3580daa793..98d157513a 100644
---
a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFile.java
+++
b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFile.java
@@ -21,13 +21,13 @@ import java.io.IOException;
import java.io.PrintWriter;
import java.io.RandomAccessFile;
import java.io.StringWriter;
+import java.lang.invoke.MethodHandles;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.FileChannel;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicInteger;
import io.netty.buffer.ByteBuf;
@@ -45,7 +45,6 @@ import
org.apache.activemq.artemis.journal.ActiveMQJournalBundle;
import org.apache.activemq.artemis.utils.Env;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.lang.invoke.MethodHandles;
public class NIOSequentialFile extends AbstractSequentialFile {
@@ -73,9 +72,8 @@ public class NIOSequentialFile extends AbstractSequentialFile
{
public NIOSequentialFile(final SequentialFileFactory factory,
final File directory,
final String file,
- final int maxIO,
- final Executor writerExecutor) {
- super(directory, file, factory, writerExecutor);
+ final int maxIO) {
+ super(directory, file, factory);
this.maxIO = maxIO;
}
@@ -375,7 +373,7 @@ public class NIOSequentialFile extends
AbstractSequentialFile {
@Override
public SequentialFile cloneFile() {
- return new NIOSequentialFile(factory, directory, getFileName(), maxIO,
null);
+ return new NIOSequentialFile(factory, directory, getFileName(), maxIO);
}
@Override
diff --git
a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFileFactory.java
b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFileFactory.java
index 8674a5767b..ead344b8f0 100644
---
a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFileFactory.java
+++
b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFileFactory.java
@@ -116,7 +116,7 @@ public class NIOSequentialFileFactory extends
AbstractSequentialFileFactory {
@Override
public SequentialFile createSequentialFile(final String fileName) {
- return new NIOSequentialFile(this, journalDir, fileName, maxIO,
writeExecutor);
+ return new NIOSequentialFile(this, journalDir, fileName, maxIO);
}
@Override
diff --git
a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalImpl.java
b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalImpl.java
index 0933f40290..ecb8dd6694 100644
---
a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalImpl.java
+++
b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalImpl.java
@@ -2836,7 +2836,7 @@ public class JournalImpl extends JournalBase implements
TestableJournal, Journal
}
if (providedIOThreadPool == null) {
- ThreadFactory factory =
AccessController.doPrivileged((PrivilegedAction<ThreadFactory>) () -> new
ActiveMQThreadFactory("ArtemisIOThread", true,
JournalImpl.class.getClassLoader()));
+ ThreadFactory factory =
AccessController.doPrivileged((PrivilegedAction<ThreadFactory>) () -> new
ActiveMQThreadFactory("io", true, JournalImpl.class.getClassLoader()));
threadPool = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L,
TimeUnit.SECONDS, new SynchronousQueue(), factory);
ioExecutorFactory = new OrderedExecutorFactory(threadPool);
diff --git
a/artemis-ra/src/main/java/org/apache/activemq/artemis/ra/inflow/ActiveMQActivation.java
b/artemis-ra/src/main/java/org/apache/activemq/artemis/ra/inflow/ActiveMQActivation.java
index f1ff7f37e4..cb178964df 100644
---
a/artemis-ra/src/main/java/org/apache/activemq/artemis/ra/inflow/ActiveMQActivation.java
+++
b/artemis-ra/src/main/java/org/apache/activemq/artemis/ra/inflow/ActiveMQActivation.java
@@ -351,7 +351,7 @@ public class ActiveMQActivation {
}
};
- Thread threadTearDown = startThread("TearDown/HornetQActivation",
runTearDown);
+ Thread threadTearDown =
startThread("resource-adapter-activation-teardown", runTearDown);
try {
threadTearDown.join(timeout);
diff --git
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/MetricsConfiguration.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/MetricsConfiguration.java
index 1ca49ed575..76df526be5 100644
---
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/MetricsConfiguration.java
+++
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/MetricsConfiguration.java
@@ -32,6 +32,7 @@ public class MetricsConfiguration implements Serializable {
private boolean uptime =
ActiveMQDefaultConfiguration.getDefaultUptimeMetrics();
private boolean logging =
ActiveMQDefaultConfiguration.getDefaultLoggingMetrics();
private boolean securityCaches =
ActiveMQDefaultConfiguration.getDefaultSecurityCacheMetrics();
+ private boolean executorServices =
ActiveMQDefaultConfiguration.getDefaultExecutorServiceMetrics();
private ActiveMQMetricsPlugin plugin;
public boolean isJvmMemory() {
@@ -123,4 +124,13 @@ public class MetricsConfiguration implements Serializable {
this.securityCaches = securityCaches;
return this;
}
+
+ public boolean isExecutorServices() {
+ return executorServices;
+ }
+
+ public MetricsConfiguration setExecutorServices(boolean executorServices) {
+ this.executorServices = executorServices;
+ return this;
+ }
}
diff --git
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java
index 264be223aa..8f100cf59a 100644
---
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java
+++
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java
@@ -1053,6 +1053,8 @@ public final class FileConfigurationParser extends
XMLConfigurationUtil {
metricsConfiguration.setLogging(XMLUtil.parseBoolean(child));
} else if (child.getNodeName().equals("security-caches")) {
metricsConfiguration.setSecurityCaches(XMLUtil.parseBoolean(child));
+ } else if (child.getNodeName().equals("executor-services")) {
+
metricsConfiguration.setExecutorServices(XMLUtil.parseBoolean(child));
} else if (child.getNodeName().equals("plugin")) {
metricsConfiguration.setPlugin(parseMetricsPlugin(child,
config));
}
diff --git
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMAcceptorFactory.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMAcceptorFactory.java
index 09592b43f2..63a4431ad0 100644
---
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMAcceptorFactory.java
+++
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMAcceptorFactory.java
@@ -21,6 +21,7 @@ import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import org.apache.activemq.artemis.core.server.cluster.ClusterConnection;
+import org.apache.activemq.artemis.core.server.metrics.MetricsManager;
import org.apache.activemq.artemis.spi.core.protocol.ProtocolManager;
import org.apache.activemq.artemis.spi.core.remoting.Acceptor;
import org.apache.activemq.artemis.spi.core.remoting.AcceptorFactory;
@@ -37,7 +38,9 @@ public class InVMAcceptorFactory implements AcceptorFactory {
final ServerConnectionLifeCycleListener
listener,
final Executor threadPool,
final ScheduledExecutorService
scheduledThreadPool,
- final Map<String, ProtocolManager>
protocolMap) {
+ final Map<String, ProtocolManager>
protocolMap,
+ String threadFactoryGroupName,
+ MetricsManager metricsManager) {
return new InVMAcceptor(name, clusterConnection, configuration, handler,
listener, protocolMap, threadPool);
}
diff --git
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java
index 6a089813ec..03af660847 100644
---
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java
+++
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java
@@ -36,6 +36,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -74,6 +75,7 @@ import org.apache.activemq.artemis.api.core.Pair;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.TransportConfiguration;
import org.apache.activemq.artemis.api.core.management.CoreNotificationType;
+import org.apache.activemq.artemis.api.core.management.ResourceNames;
import org.apache.activemq.artemis.core.client.impl.ClientSessionFactoryImpl;
import org.apache.activemq.artemis.core.protocol.ProtocolHandler;
import org.apache.activemq.artemis.core.remoting.impl.AbstractAcceptor;
@@ -85,6 +87,7 @@ import
org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
import org.apache.activemq.artemis.core.server.cluster.ClusterConnection;
import org.apache.activemq.artemis.core.server.management.Notification;
import org.apache.activemq.artemis.core.server.management.NotificationService;
+import org.apache.activemq.artemis.core.server.metrics.MetricsManager;
import org.apache.activemq.artemis.spi.core.protocol.ProtocolManager;
import org.apache.activemq.artemis.spi.core.remoting.BufferHandler;
import org.apache.activemq.artemis.spi.core.remoting.Connection;
@@ -251,6 +254,10 @@ public class NettyAcceptor extends AbstractAcceptor {
private volatile int actualPort = 0;
+ private final String threadFactoryGroupName;
+
+ private final MetricsManager metricsManager;
+
public NettyAcceptor(final String name,
final ClusterConnection clusterConnection,
final Map<String, Object> configuration,
@@ -258,7 +265,9 @@ public class NettyAcceptor extends AbstractAcceptor {
final ServerConnectionLifeCycleListener listener,
final ScheduledExecutorService scheduledThreadPool,
final Executor failureExecutor,
- final Map<String, ProtocolManager> protocolMap) {
+ final Map<String, ProtocolManager> protocolMap,
+ final String threadFactoryGroupName,
+ final MetricsManager metricsManager) {
super(protocolMap);
this.failureExecutor = failureExecutor;
@@ -273,9 +282,13 @@ public class NettyAcceptor extends AbstractAcceptor {
this.listener = listener;
+ this.threadFactoryGroupName = threadFactoryGroupName;
+
+ this.metricsManager = metricsManager;
+
sslEnabled =
ConfigurationHelper.getBooleanProperty(TransportConstants.SSL_ENABLED_PROP_NAME,
TransportConstants.DEFAULT_SSL_ENABLED, configuration);
- remotingThreads =
ConfigurationHelper.getIntProperty(TransportConstants.NIO_REMOTING_THREADS_PROPNAME,
-1, configuration);
+ remotingThreads =
ConfigurationHelper.getIntProperty(TransportConstants.NIO_REMOTING_THREADS_PROPNAME,
Runtime.getRuntime().availableProcessors() * 3, configuration);
remotingThreads =
ConfigurationHelper.getIntProperty(TransportConstants.REMOTING_THREADS_PROPNAME,
remotingThreads, configuration);
useEpoll =
ConfigurationHelper.getBooleanProperty(TransportConstants.USE_EPOLL_PROP_NAME,
TransportConstants.DEFAULT_USE_EPOLL, configuration);
@@ -428,32 +441,29 @@ public class NettyAcceptor extends AbstractAcceptor {
channelClazz = LocalServerChannel.class;
eventLoopGroup = new DefaultEventLoopGroup();
} else {
-
- if (remotingThreads == -1) {
- // Default to number of cores * 3
- remotingThreads = Runtime.getRuntime().availableProcessors() * 3;
- }
-
+ ThreadFactory threadFactory =
AccessController.doPrivileged((PrivilegedAction<ActiveMQThreadFactory>) () ->
new ActiveMQThreadFactory(threadFactoryGroupName, true,
ClientSessionFactoryImpl.class.getClassLoader()));
if (useEpoll && CheckDependencies.isEpollAvailable()) {
channelClazz = EpollServerSocketChannel.class;
- eventLoopGroup = new EpollEventLoopGroup(remotingThreads,
AccessController.doPrivileged((PrivilegedAction<ActiveMQThreadFactory>) () ->
new ActiveMQThreadFactory("activemq-netty-threads", true,
ClientSessionFactoryImpl.class.getClassLoader())));
+ eventLoopGroup = new EpollEventLoopGroup(remotingThreads,
threadFactory);
acceptorType = EPOLL_ACCEPTOR_TYPE;
-
- logger.debug("Acceptor using native epoll");
+ logger.debug("Acceptor {} using native epoll", name);
} else if (useKQueue && CheckDependencies.isKQueueAvailable()) {
channelClazz = KQueueServerSocketChannel.class;
- eventLoopGroup = new KQueueEventLoopGroup(remotingThreads,
AccessController.doPrivileged((PrivilegedAction<ActiveMQThreadFactory>) () ->
new ActiveMQThreadFactory("activemq-netty-threads", true,
ClientSessionFactoryImpl.class.getClassLoader())));
+ eventLoopGroup = new KQueueEventLoopGroup(remotingThreads,
threadFactory);
acceptorType = KQUEUE_ACCEPTOR_TYPE;
-
- logger.debug("Acceptor using native kqueue");
+ logger.debug("Acceptor {} using native kqueue", name);
} else {
channelClazz = NioServerSocketChannel.class;
- eventLoopGroup = new NioEventLoopGroup(remotingThreads,
AccessController.doPrivileged((PrivilegedAction<ActiveMQThreadFactory>) () ->
new ActiveMQThreadFactory("activemq-netty-threads", true,
ClientSessionFactoryImpl.class.getClassLoader())));
+ eventLoopGroup = new NioEventLoopGroup(remotingThreads,
threadFactory);
acceptorType = NIO_ACCEPTOR_TYPE;
- logger.debug("Acceptor using nio");
+ logger.debug("Acceptor {} using nio", name);
}
}
+ if (metricsManager != null) {
+ metricsManager.registerNettyEventLoopGroup(name, eventLoopGroup);
+ }
+
bootstrap = new ServerBootstrap();
bootstrap.group(eventLoopGroup);
bootstrap.channel(channelClazz);
@@ -736,10 +746,11 @@ public class NettyAcceptor extends AbstractAcceptor {
@Override
public void stop() throws Exception {
CountDownLatch latch = new CountDownLatch(1);
-
asyncStop(latch::countDown);
-
latch.await();
+ if (metricsManager != null) {
+ metricsManager.remove(ResourceNames.ACCEPTOR + this.name);
+ }
}
@Override
@@ -1032,6 +1043,7 @@ public class NettyAcceptor extends AbstractAcceptor {
}
}
+ @Override
public boolean isAutoStart() {
return autoStart;
}
@@ -1040,4 +1052,8 @@ public class NettyAcceptor extends AbstractAcceptor {
public int getActualPort() {
return actualPort;
}
+
+ public int getRemotingThreads() {
+ return remotingThreads;
+ }
}
diff --git
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptorFactory.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptorFactory.java
index dbaa731854..64c6087bea 100644
---
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptorFactory.java
+++
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptorFactory.java
@@ -21,6 +21,7 @@ import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import org.apache.activemq.artemis.core.server.cluster.ClusterConnection;
+import org.apache.activemq.artemis.core.server.metrics.MetricsManager;
import org.apache.activemq.artemis.spi.core.protocol.ProtocolManager;
import org.apache.activemq.artemis.spi.core.remoting.Acceptor;
import org.apache.activemq.artemis.spi.core.remoting.AcceptorFactory;
@@ -38,8 +39,10 @@ public class NettyAcceptorFactory implements AcceptorFactory
{
final ServerConnectionLifeCycleListener
listener,
final Executor threadPool,
final ScheduledExecutorService
scheduledThreadPool,
- final Map<String, ProtocolManager>
protocolMap) {
+ final Map<String, ProtocolManager>
protocolMap,
+ final String threadFactoryGroupName,
+ MetricsManager metricsManager) {
Executor failureExecutor = new OrderedExecutor(threadPool);
- return new NettyAcceptor(name, connection, configuration, handler,
listener, scheduledThreadPool, failureExecutor, protocolMap);
+ return new NettyAcceptor(name, connection, configuration, handler,
listener, scheduledThreadPool, failureExecutor, protocolMap,
threadFactoryGroupName, metricsManager);
}
}
diff --git
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/server/impl/RemotingServiceImpl.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/server/impl/RemotingServiceImpl.java
index 25907935ae..207b869d61 100644
---
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/server/impl/RemotingServiceImpl.java
+++
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/server/impl/RemotingServiceImpl.java
@@ -16,13 +16,6 @@
*/
package org.apache.activemq.artemis.core.remoting.server.impl;
-import static
org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.DEFAULT_SSL_AUTO_RELOAD;
-import static
org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.KEYSTORE_PATH_PROP_NAME;
-import static
org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.KEYSTORE_TYPE_PROP_NAME;
-import static
org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.SSL_AUTO_RELOAD_PROP_NAME;
-import static
org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.TRUSTSTORE_PATH_PROP_NAME;
-import static
org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.TRUSTSTORE_TYPE_PROP_NAME;
-
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
@@ -69,7 +62,6 @@ import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.ConfigurationUtils;
import org.apache.activemq.artemis.core.protocol.core.CoreRemotingConnection;
import
org.apache.activemq.artemis.core.protocol.core.impl.CoreProtocolManagerFactory;
-import org.apache.activemq.artemis.core.remoting.impl.netty.NettyAcceptor;
import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants;
import org.apache.activemq.artemis.core.remoting.server.RemotingService;
import org.apache.activemq.artemis.core.security.ActiveMQPrincipal;
@@ -103,6 +95,13 @@ import org.apache.activemq.artemis.utils.ReusableLatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static
org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.DEFAULT_SSL_AUTO_RELOAD;
+import static
org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.KEYSTORE_PATH_PROP_NAME;
+import static
org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.KEYSTORE_TYPE_PROP_NAME;
+import static
org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.SSL_AUTO_RELOAD_PROP_NAME;
+import static
org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.TRUSTSTORE_PATH_PROP_NAME;
+import static
org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.TRUSTSTORE_TYPE_PROP_NAME;
+
public class RemotingServiceImpl implements RemotingService,
ServerConnectionLifeCycleListener {
private static final Logger logger =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@@ -223,15 +222,8 @@ public class RemotingServiceImpl implements
RemotingService, ServerConnectionLif
paused = false;
- // The remoting service maintains it's own thread pool for handling
remoting traffic
- // If OIO each connection will have it's own thread
- // If NIO these are capped at nio-remoting-threads which defaults to num
cores * 3
- // This needs to be a different thread pool to the main thread pool
especially for OIO where we may need
- // to support many hundreds of connections, but the main thread pool
must be kept small for better performance
-
- ThreadFactory tFactory =
AccessController.doPrivileged((PrivilegedAction<ThreadFactory>) () -> new
ActiveMQThreadFactory("ActiveMQ-remoting-threads-" + server.toString() + "-" +
System.identityHashCode(this), false,
Thread.currentThread().getContextClassLoader()));
-
- threadPool = Executors.newCachedThreadPool(tFactory);
+ // this is used for in-vm but for Netty it's only used for executing
failure listeners
+ threadPool =
Executors.newCachedThreadPool(AccessController.doPrivileged((PrivilegedAction<ThreadFactory>)
() -> new ActiveMQThreadFactory(server.getThreadGroupName("remoting-service"),
false, Thread.currentThread().getContextClassLoader())));
for (TransportConfiguration info : acceptorsConfig.values()) {
createAcceptor(info);
@@ -291,7 +283,7 @@ public class RemotingServiceImpl implements
RemotingService, ServerConnectionLif
selectedProtocols.put(entry.getKey(),
entry.getValue().createProtocolManager(server, info.getCombinedParams(),
incomingInterceptors, outgoingInterceptors));
}
- acceptor = factory.createAcceptor(info.getName(), clusterConnection,
info.getParams(), new DelegatingBufferHandler(), this, threadPool,
scheduledThreadPool, selectedProtocols);
+ acceptor = factory.createAcceptor(info.getName(), clusterConnection,
info.getParams(), new DelegatingBufferHandler(), this, threadPool,
scheduledThreadPool, selectedProtocols, server.getThreadGroupName("remoting-" +
info.getName()), server.getMetricsManager());
if (defaultInvmSecurityPrincipal != null && acceptor.isUnsecurable())
{
acceptor.setDefaultActiveMQPrincipal(defaultInvmSecurityPrincipal);
@@ -345,7 +337,7 @@ public class RemotingServiceImpl implements
RemotingService, ServerConnectionLif
if (isStarted()) {
for (Acceptor a : acceptors.values()) {
try {
- if (a instanceof NettyAcceptor acceptor &&
!acceptor.isAutoStart()) {
+ if (!a.isAutoStart()) {
continue;
}
a.start();
@@ -707,8 +699,8 @@ public class RemotingServiceImpl implements
RemotingService, ServerConnectionLif
for (TransportConfiguration candidateConfiguration : acceptorsToCreate) {
final Acceptor acceptor = createAcceptor(candidateConfiguration);
- if (isStarted() && acceptor instanceof NettyAcceptor startable &&
startable.isAutoStart()) {
- acceptorsToStart.add(startable);
+ if (isStarted() && acceptor.isAutoStart()) {
+ acceptorsToStart.add(acceptor);
}
}
diff --git
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
index 21fa5b87aa..47e645ac13 100644
---
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
+++
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
@@ -153,6 +153,8 @@ public interface ActiveMQServer extends ServiceComponent {
void removeMirrorControl();
+ String getThreadGroupName(String groupName);
+
ServiceRegistry getServiceRegistry();
RemotingService getRemotingService();
diff --git
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
index 9cc142d204..a39508fe9b 100644
---
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
+++
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
@@ -40,6 +40,7 @@ import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -165,6 +166,7 @@ import
org.apache.activemq.artemis.core.server.group.impl.RemoteGroupingHandler;
import org.apache.activemq.artemis.core.server.impl.jdbc.JdbcNodeManager;
import org.apache.activemq.artemis.core.server.management.ManagementService;
import
org.apache.activemq.artemis.core.server.management.impl.ManagementServiceImpl;
+import org.apache.activemq.artemis.core.server.metrics.BrokerMetricNames;
import org.apache.activemq.artemis.core.server.metrics.MetricsManager;
import org.apache.activemq.artemis.core.server.mirror.MirrorController;
import org.apache.activemq.artemis.core.server.mirror.MirrorRegistry;
@@ -249,6 +251,8 @@ public class ActiveMQServerImpl implements ActiveMQServer {
@Deprecated
public static final String GENERIC_IGNORED_FILTER =
Filter.GENERIC_IGNORED_FILTER;
+ private static final long THREAD_POOL_KEEP_ALIVE_SECONDS = 60L;
+
private HAPolicy haPolicy;
private MirrorRegistry mirrorRegistry = new MirrorRegistry();
@@ -1024,7 +1028,7 @@ public class ActiveMQServerImpl implements ActiveMQServer
{
}
@Override
- public ExecutorService getThreadPool() {
+ public Executor getThreadPool() {
return threadPool;
}
@@ -3190,17 +3194,17 @@ public class ActiveMQServerImpl implements
ActiveMQServer {
* Sets up ActiveMQ Artemis Executor Services.
*/
private void initializeExecutorServices() {
+ int maxIoThreads = configuration.getPageMaxConcurrentIO() <= 0 ?
Integer.MAX_VALUE : configuration.getPageMaxConcurrentIO();
/*
* We check to see if a Thread Pool is supplied in the
InjectedObjectRegistry. If so we created a new Ordered
* Executor based on the provided Thread pool. Otherwise we create a
new ThreadPool.
*/
if (serviceRegistry.getExecutorService() == null) {
- ThreadFactory tFactory =
AccessController.doPrivileged((PrivilegedAction<ThreadFactory>) ()-> new
ActiveMQThreadFactory("ActiveMQ-server-" + this, false,
ClientSessionFactoryImpl.class.getClassLoader()));
-
+ ThreadFactory tFactory = getThreadFactory(null);
if (configuration.getThreadPoolMaxSize() == -1) {
- threadPool = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L,
TimeUnit.SECONDS, new SynchronousQueue<>(), tFactory);
+ threadPool = new ThreadPoolExecutor(0, Integer.MAX_VALUE,
THREAD_POOL_KEEP_ALIVE_SECONDS, TimeUnit.SECONDS, new SynchronousQueue<>(),
tFactory);
} else {
- threadPool = new ActiveMQThreadPoolExecutor(0,
configuration.getThreadPoolMaxSize(), 60L, TimeUnit.SECONDS, tFactory);
+ threadPool = new ActiveMQThreadPoolExecutor(0,
configuration.getThreadPoolMaxSize(), THREAD_POOL_KEEP_ALIVE_SECONDS,
TimeUnit.SECONDS, tFactory);
}
} else {
threadPool = serviceRegistry.getExecutorService();
@@ -3208,40 +3212,26 @@ public class ActiveMQServerImpl implements
ActiveMQServer {
}
this.executorFactory = new OrderedExecutorFactory(threadPool);
- if (serviceRegistry.getIOExecutorService() != null) {
- this.ioExecutorFactory = new
OrderedExecutorFactory(serviceRegistry.getIOExecutorService());
+ if (serviceRegistry.getIOExecutorService() == null) {
+ this.ioExecutorPool = new ActiveMQThreadPoolExecutor(0, maxIoThreads,
THREAD_POOL_KEEP_ALIVE_SECONDS, TimeUnit.SECONDS, getThreadFactory("io"));
} else {
- ThreadFactory tFactory =
AccessController.doPrivileged((PrivilegedAction<ThreadFactory>) () -> new
ActiveMQThreadFactory("ActiveMQ-IO-server-" + this, false,
ClientSessionFactoryImpl.class.getClassLoader()));
-
- // Perhaps getPageMaxConcurrentIO should be deprecated and a new
value added
- int maxIO = configuration.getPageMaxConcurrentIO() <= 0 ?
Integer.MAX_VALUE : configuration.getPageMaxConcurrentIO();
- this.ioExecutorPool = new ActiveMQThreadPoolExecutor(0, maxIO, 60L,
TimeUnit.SECONDS, tFactory);
- this.ioExecutorFactory = new OrderedExecutorFactory(ioExecutorPool);
+ this.ioExecutorPool = serviceRegistry.getIOExecutorService();
}
+ this.ioExecutorFactory = new OrderedExecutorFactory(ioExecutorPool);
- if (serviceRegistry.getPageExecutorService() != null) {
- this.pageExecutorFactory = new
OrderedExecutorFactory(serviceRegistry.getPageExecutorService()).setFair(true);
+ if (serviceRegistry.getPageExecutorService() == null) {
+ this.pageExecutorPool = new ActiveMQThreadPoolExecutor(0,
maxIoThreads, THREAD_POOL_KEEP_ALIVE_SECONDS, TimeUnit.SECONDS,
getThreadFactory("paging"));
} else {
- ThreadFactory tFactory = AccessController.doPrivileged(new
PrivilegedAction<>() {
- @Override
- public ThreadFactory run() {
- return new
ActiveMQThreadFactory("ActiveMQ-PageExecutor-server-" + this.toString(), false,
ClientSessionFactoryImpl.class.getClassLoader());
- }
- });
-
- int maxIO = configuration.getPageMaxConcurrentIO() <= 0 ?
Integer.MAX_VALUE : configuration.getPageMaxConcurrentIO();
- this.pageExecutorPool = new ActiveMQThreadPoolExecutor(0, maxIO, 60L,
TimeUnit.SECONDS, tFactory);
- this.pageExecutorFactory = new
OrderedExecutorFactory(pageExecutorPool);
+ this.pageExecutorPool = serviceRegistry.getPageExecutorService();
}
+ this.pageExecutorFactory = new OrderedExecutorFactory(pageExecutorPool);
- /*
- * We check to see if a Scheduled Executor Service is provided in the
InjectedObjectRegistry. If so we use this
+ /*
+ * We check to see if a Scheduled Executor Service is provided in the
InjectedObjectRegistry. If so we use this
* Scheduled ExecutorService otherwise we create a new one.
*/
if (serviceRegistry.getScheduledExecutorService() == null) {
- ThreadFactory tFactory =
AccessController.doPrivileged((PrivilegedAction<ThreadFactory>) () -> new
ActiveMQThreadFactory("ActiveMQ-scheduled-threads", false,
ClientSessionFactoryImpl.class.getClassLoader()));
-
- ScheduledThreadPoolExecutor scheduledPoolExecutor = new
ScheduledThreadPoolExecutor(configuration.getScheduledThreadPoolMaxSize(),
tFactory);
+ ScheduledThreadPoolExecutor scheduledPoolExecutor = new
ScheduledThreadPoolExecutor(configuration.getScheduledThreadPoolMaxSize(),
getThreadFactory("scheduled"));
scheduledPoolExecutor.setRemoveOnCancelPolicy(true);
scheduledPool = scheduledPoolExecutor;
} else {
@@ -3255,6 +3245,23 @@ public class ActiveMQServerImpl implements
ActiveMQServer {
}
}
+ private ActiveMQThreadFactory getThreadFactory(String name) {
+ return
AccessController.doPrivileged((PrivilegedAction<ActiveMQThreadFactory>) () ->
new ActiveMQThreadFactory(getThreadGroupName(name), false,
ClientSessionFactoryImpl.class.getClassLoader()));
+ }
+
+ @Override
+ public String getThreadGroupName(String groupName) {
+ final String tFactoryGroupNameSuffix;
+ if (identity != null) {
+ tFactoryGroupNameSuffix = identity;
+ } else if (configuration != null && configuration.getName() != null &&
!configuration.getName().isEmpty()) {
+ tFactoryGroupNameSuffix = configuration.getName();
+ } else {
+ tFactoryGroupNameSuffix =
Integer.toHexString(System.identityHashCode(this));
+ }
+ return (groupName == null ? "" : groupName + "-") +
tFactoryGroupNameSuffix;
+ }
+
@Override
public ServiceRegistry getServiceRegistry() {
return serviceRegistry;
@@ -3317,6 +3324,10 @@ public class ActiveMQServerImpl implements
ActiveMQServer {
*/
if (configuration.getMetricsConfiguration() != null &&
configuration.getMetricsConfiguration().getPlugin() != null) {
metricsManager = new MetricsManager(configuration.getName(),
configuration.getMetricsConfiguration(), addressSettingsRepository,
securityStore);
+
metricsManager.registerExecutorService(BrokerMetricNames.GENERAL_EXECUTOR_SERVICE,
threadPool);
+
metricsManager.registerExecutorService(BrokerMetricNames.IO_EXECUTOR_SERVICE,
ioExecutorPool);
+
metricsManager.registerExecutorService(BrokerMetricNames.PAGE_EXECUTOR_SERVICE,
pageExecutorPool);
+
metricsManager.registerExecutorService(BrokerMetricNames.SCHEDULED_EXECUTOR_SERVICE,
scheduledPool);
}
postOffice = new PostOfficeImpl(this, storageManager, pagingManager,
queueFactory, managementService, configuration.getMessageExpiryScanPeriod(),
configuration.getAddressQueueScanPeriod(),
configuration.getWildcardConfiguration(), configuration.getIDCacheSize(),
configuration.isPersistIDCache(), addressSettingsRepository);
diff --git
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ReplicationPrimaryActivation.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ReplicationPrimaryActivation.java
index 9d683d7c82..fc1a3a7d88 100644
---
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ReplicationPrimaryActivation.java
+++
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ReplicationPrimaryActivation.java
@@ -17,8 +17,9 @@
package org.apache.activemq.artemis.core.server.impl;
import javax.annotation.concurrent.GuardedBy;
+import java.lang.invoke.MethodHandles;
import java.util.Objects;
-import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -49,7 +50,6 @@ import
org.apache.activemq.artemis.lockmanager.UnavailableStateException;
import org.apache.activemq.artemis.spi.core.remoting.Acceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.lang.invoke.MethodHandles;
import static
org.apache.activemq.artemis.core.server.ActiveMQServer.SERVER_STATE.STARTED;
import static
org.apache.activemq.artemis.core.server.impl.quorum.ActivationSequenceStateMachine.awaitNextCommittedActivationSequence;
@@ -374,12 +374,12 @@ public class ReplicationPrimaryActivation extends
PrimaryActivation implements D
}
private void onReplicationConnectionClose() {
- ExecutorService executorService = activeMQServer.getThreadPool();
- if (executorService != null) {
+ Executor executor = activeMQServer.getThreadPool();
+ if (executor != null) {
if (stoppingServer.get()) {
return;
}
- executorService.execute(() -> {
+ executor.execute(() -> {
synchronized (replicationLock) {
if (replicationManager == null) {
return;
diff --git
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingPrimaryActivation.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingPrimaryActivation.java
index 8a8848a102..bc1e63be39 100644
---
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingPrimaryActivation.java
+++
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingPrimaryActivation.java
@@ -16,9 +16,10 @@
*/
package org.apache.activemq.artemis.core.server.impl;
+import java.lang.invoke.MethodHandles;
import java.util.List;
import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import
org.apache.activemq.artemis.api.core.ActiveMQAlreadyReplicatingException;
@@ -61,7 +62,6 @@ import
org.apache.activemq.artemis.core.server.cluster.quorum.QuorumManager;
import org.apache.activemq.artemis.spi.core.remoting.Acceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.lang.invoke.MethodHandles;
public class SharedNothingPrimaryActivation extends PrimaryActivation {
@@ -243,9 +243,9 @@ public class SharedNothingPrimaryActivation extends
PrimaryActivation {
}
private void handleClose(boolean failed) {
- ExecutorService executorService = activeMQServer.getThreadPool();
- if (executorService != null) {
- executorService.execute(() -> {
+ Executor executor = activeMQServer.getThreadPool();
+ if (executor != null) {
+ executor.execute(() -> {
synchronized (replicationLock) {
if (replicationManager != null) {
activeMQServer.getStorageManager().stopReplication();
diff --git
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/metrics/BrokerMetricNames.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/metrics/BrokerMetricNames.java
index dc31fce2c6..53994712e4 100644
---
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/metrics/BrokerMetricNames.java
+++
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/metrics/BrokerMetricNames.java
@@ -29,4 +29,8 @@ public class BrokerMetricNames {
public static final String ACTIVE = "active";
public static final String AUTHENTICATION_COUNT = "authentication.count";
public static final String AUTHORIZATION_COUNT = "authorization.count";
+ public static final String GENERAL_EXECUTOR_SERVICE =
"general.executor.service";
+ public static final String IO_EXECUTOR_SERVICE = "io.executor.service";
+ public static final String PAGE_EXECUTOR_SERVICE =
"paging.executor.service";
+ public static final String SCHEDULED_EXECUTOR_SERVICE =
"scheduled.executor.service";
}
diff --git
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/metrics/MetricsManager.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/metrics/MetricsManager.java
index d3ed24eeb1..9d5f32d08f 100644
---
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/metrics/MetricsManager.java
+++
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/metrics/MetricsManager.java
@@ -19,9 +19,13 @@ package org.apache.activemq.artemis.core.server.metrics;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
import java.util.function.Consumer;
import java.util.function.ToDoubleFunction;
@@ -33,14 +37,19 @@ import io.micrometer.core.instrument.Metrics;
import io.micrometer.core.instrument.Tag;
import io.micrometer.core.instrument.Tags;
import io.micrometer.core.instrument.binder.cache.CaffeineCacheMetrics;
+import io.micrometer.core.instrument.binder.jvm.ExecutorServiceMetrics;
import io.micrometer.core.instrument.binder.jvm.JvmGcMetrics;
import io.micrometer.core.instrument.binder.jvm.JvmMemoryMetrics;
import io.micrometer.core.instrument.binder.jvm.JvmThreadMetrics;
import io.micrometer.core.instrument.binder.logging.Log4j2Metrics;
+import io.micrometer.core.instrument.binder.netty4.NettyMeters;
import io.micrometer.core.instrument.binder.system.FileDescriptorMetrics;
import io.micrometer.core.instrument.binder.system.ProcessorMetrics;
import io.micrometer.core.instrument.binder.system.UptimeMetrics;
+import io.micrometer.core.instrument.config.MeterFilter;
import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.channel.EventLoopGroup;
+import io.netty.util.concurrent.SingleThreadEventExecutor;
import org.apache.activemq.artemis.api.core.management.ResourceNames;
import org.apache.activemq.artemis.core.config.MetricsConfiguration;
import org.apache.activemq.artemis.core.security.SecurityStore;
@@ -51,9 +60,12 @@ import
org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static
io.micrometer.core.instrument.binder.netty4.NettyMeters.EVENT_EXECUTOR_TASKS_PENDING;
+
public class MetricsManager {
public static final String BROKER_TAG_NAME = "broker";
+ public static final String METER_PREFIX = "artemis.";
private static final Logger logger =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@@ -62,19 +74,34 @@ public class MetricsManager {
private final Tags commonTags;
private final MeterRegistry meterRegistry;
- private final Map<String, List<Meter>> meters = new ConcurrentHashMap<>();
+ private final Map<String, Collection<Meter>> meters = new
ConcurrentHashMap<>();
private final HierarchicalRepository<AddressSettings>
addressSettingsRepository;
+ private final MetricsConfiguration metricsConfiguration;
+
public MetricsManager(String brokerName,
MetricsConfiguration metricsConfiguration,
HierarchicalRepository<AddressSettings>
addressSettingsRepository,
SecurityStore securityStore) {
this.brokerName = brokerName;
+ this.metricsConfiguration = metricsConfiguration;
this.meterRegistry = metricsConfiguration.getPlugin().getRegistry();
this.addressSettingsRepository = addressSettingsRepository;
this.commonTags = Tags.of(BROKER_TAG_NAME, brokerName);
if (meterRegistry != null) {
+
+ // This is a temporary workaround until Micrometer supports common
tags on NettyEventExecutorMetrics
+ meterRegistry.config().meterFilter(new MeterFilter() {
+ @Override
+ public Meter.Id map(Meter.Id id) {
+ if
(id.getName().equals(EVENT_EXECUTOR_TASKS_PENDING.getName())) {
+ return id.withTags(commonTags);
+ }
+ return id;
+ }
+ });
+
Metrics.globalRegistry.add(meterRegistry);
if (metricsConfiguration.isJvmMemory()) {
new JvmMemoryMetrics(commonTags).bindTo(meterRegistry);
@@ -124,7 +151,7 @@ public class MetricsManager {
final List<Builder<Object>> gaugeBuilders = new ArrayList<>();
builder.accept((metricName, state, f, description, gaugeTags) -> {
Builder<Object> meter = Gauge
- .builder("artemis." + metricName, state, f)
+ .builder(METER_PREFIX + metricName, state, f)
.tags(commonTags)
.tags(gaugeTags)
.tag("address", address)
@@ -142,7 +169,7 @@ public class MetricsManager {
final List<Builder<Object>> gaugeBuilders = new ArrayList<>();
builder.accept((metricName, state, f, description, gaugeTags) -> {
Builder<Object> meter = Gauge
- .builder("artemis." + metricName, state, f)
+ .builder(METER_PREFIX + metricName, state, f)
.tags(commonTags)
.tags(gaugeTags)
.tag("address", address)
@@ -159,7 +186,7 @@ public class MetricsManager {
final List<Builder<Object>> gaugeBuilders = new ArrayList<>();
builder.accept((metricName, state, f, description, gaugeTags) -> {
Builder<Object> meter = Gauge
- .builder("artemis." + metricName, state, f)
+ .builder(METER_PREFIX + metricName, state, f)
.tags(commonTags)
.tags(gaugeTags)
.description(description);
@@ -173,7 +200,7 @@ public class MetricsManager {
throw ActiveMQMessageBundle.BUNDLE.metersAlreadyRegistered(resource);
}
logger.debug("Registering meters for {}", resource);
- List<Meter> newMeters = new ArrayList<>(gaugeBuilders.size());
+ Set<Meter> newMeters = new HashSet<>(gaugeBuilders.size());
for (Builder<Object> gaugeBuilder : gaugeBuilders) {
Gauge gauge = gaugeBuilder.register(meterRegistry);
newMeters.add(gauge);
@@ -183,7 +210,7 @@ public class MetricsManager {
}
public void remove(String resource) {
- List<Meter> resourceMeters = meters.remove(resource);
+ Collection<Meter> resourceMeters = meters.remove(resource);
if (resourceMeters != null) {
logger.debug("Unregistering meters for {}", resource);
for (Meter meter : resourceMeters) {
@@ -198,4 +225,42 @@ public class MetricsManager {
logger.debug("Attempted to unregister meters for {}, but none were
found.", resource);
}
}
+
+ public void registerExecutorService(String name, ExecutorService
executorService) {
+ if (this.meterRegistry == null ||
!metricsConfiguration.isExecutorServices()) {
+ return;
+ }
+ logger.debug("Registering ExecutorService for {}: {}", name,
executorService);
+ ExecutorServiceMetrics.monitor(meterRegistry, executorService, name,
commonTags);
+ }
+
+ /* This method is based on
io.micrometer.core.instrument.binder.netty4.NettyEventExecutorMetrics but that
class
+ * doesn't have a way to track and unregister meters later which is
necessary when acceptors are stopped or
+ * destroyed.
+ */
+ public void registerNettyEventLoopGroup(String name, EventLoopGroup
eventLoopGroup) {
+ if (this.meterRegistry == null ||
!metricsConfiguration.isExecutorServices()) {
+ return;
+ }
+ String resource = ResourceNames.ACCEPTOR + name;
+ if (meters.get(resource) != null) {
+ throw ActiveMQMessageBundle.BUNDLE.metersAlreadyRegistered(resource);
+ }
+ logger.debug("Registering Netty EventLoopGroup for {}: {}", name,
eventLoopGroup);
+ Set<Meter> meters = new HashSet<>();
+ eventLoopGroup.forEach(eventExecutor -> {
+ if (eventExecutor instanceof SingleThreadEventExecutor) {
+ SingleThreadEventExecutor singleThreadEventExecutor =
(SingleThreadEventExecutor) eventExecutor;
+ Meter meter =
Gauge.builder(NettyMeters.EVENT_EXECUTOR_TASKS_PENDING.getName(),
singleThreadEventExecutor::pendingTasks)
+ .tag("name",
singleThreadEventExecutor.threadProperties().name())
+ .tags(commonTags)
+ .register(this.meterRegistry);
+ meters.add(meter);
+ logger.debug("Registered meter: {}", meter.getId());
+ }
+ });
+ if (!meters.isEmpty()) {
+ this.meters.put(resource, meters);
+ }
+ }
}
diff --git
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/metrics/NettyPooledAllocatorMetrics.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/metrics/NettyPooledAllocatorMetrics.java
index dd9e0762ce..4f07264d5c 100644
---
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/metrics/NettyPooledAllocatorMetrics.java
+++
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/metrics/NettyPooledAllocatorMetrics.java
@@ -19,16 +19,16 @@ package org.apache.activemq.artemis.core.server.metrics;
import java.util.List;
import java.util.function.Function;
+import io.micrometer.core.instrument.FunctionCounter;
import io.micrometer.core.instrument.Gauge;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Tags;
import io.micrometer.core.instrument.binder.MeterBinder;
+import io.netty.buffer.PoolArenaMetric;
import io.netty.buffer.PoolChunkListMetric;
import io.netty.buffer.PoolChunkMetric;
import io.netty.buffer.PoolSubpageMetric;
import io.netty.buffer.PooledByteBufAllocatorMetric;
-import io.micrometer.core.instrument.FunctionCounter;
-import io.netty.buffer.PoolArenaMetric;
public final class NettyPooledAllocatorMetrics implements MeterBinder {
diff --git
a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/remoting/Acceptor.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/remoting/Acceptor.java
index 2f0b08e7de..5914c8c59e 100644
---
a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/remoting/Acceptor.java
+++
b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/remoting/Acceptor.java
@@ -26,6 +26,8 @@ import
org.apache.activemq.artemis.core.server.ActiveMQComponent;
import org.apache.activemq.artemis.core.server.cluster.ClusterConnection;
import org.apache.activemq.artemis.core.server.management.NotificationService;
+import static
org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.DEFAULT_AUTO_START;
+
/**
* An Acceptor is used by the RemotingService to allow clients to connect. It
should take care of dispatching client
* requests to the RemotingService's Dispatcher.
@@ -94,4 +96,8 @@ public interface Acceptor extends ActiveMQComponent {
default int getActualPort() {
return -1;
}
+
+ default boolean isAutoStart() {
+ return DEFAULT_AUTO_START;
+ }
}
diff --git
a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/remoting/AcceptorFactory.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/remoting/AcceptorFactory.java
index b944e89ec3..e6d53f332a 100644
---
a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/remoting/AcceptorFactory.java
+++
b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/remoting/AcceptorFactory.java
@@ -21,6 +21,7 @@ import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import org.apache.activemq.artemis.core.server.cluster.ClusterConnection;
+import org.apache.activemq.artemis.core.server.metrics.MetricsManager;
import org.apache.activemq.artemis.spi.core.protocol.ProtocolManager;
/**
@@ -49,7 +50,9 @@ public interface AcceptorFactory {
ServerConnectionLifeCycleListener listener,
Executor threadPool,
ScheduledExecutorService scheduledThreadPool,
- Map<String, ProtocolManager> protocolMap);
+ Map<String, ProtocolManager> protocolMap,
+ String threadFactoryGroupName,
+ MetricsManager metricsManager);
default boolean supportsRemote() {
return true;
diff --git a/artemis-server/src/main/resources/schema/artemis-configuration.xsd
b/artemis-server/src/main/resources/schema/artemis-configuration.xsd
index c92486a4f2..5e87d2e00f 100644
--- a/artemis-server/src/main/resources/schema/artemis-configuration.xsd
+++ b/artemis-server/src/main/resources/schema/artemis-configuration.xsd
@@ -5231,6 +5231,14 @@
</xsd:annotation>
</xsd:element>
+ <xsd:element name="executor-services" type="xsd:boolean"
default="false" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ whether to report metrics for the internal executor
services (i.e. thread pools)
+ </xsd:documentation>
+ </xsd:annotation>
+ </xsd:element>
+
<xsd:element name="plugin" maxOccurs="1" minOccurs="0">
<xsd:complexType>
<xsd:annotation>
diff --git
a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/DefaultsFileConfigurationTest.java
b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/DefaultsFileConfigurationTest.java
index 5e20cf5c9a..ac82dd031a 100644
---
a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/DefaultsFileConfigurationTest.java
+++
b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/DefaultsFileConfigurationTest.java
@@ -161,5 +161,7 @@ public class DefaultsFileConfigurationTest extends
AbstractConfigurationTestBase
assertEquals(ActiveMQDefaultConfiguration.getDefaultLoggingMetrics(),
conf.getMetricsConfiguration().isLogging());
assertEquals(ActiveMQDefaultConfiguration.getDefaultSecurityCacheMetrics(),
conf.getMetricsConfiguration().isSecurityCaches());
+
+
assertEquals(ActiveMQDefaultConfiguration.getDefaultExecutorServiceMetrics(),
conf.getMetricsConfiguration().isExecutorServices());
}
}
diff --git
a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
index 24c32aaf9d..b59bc1ccbb 100644
---
a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
+++
b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
@@ -685,6 +685,7 @@ public class FileConfigurationTest extends
AbstractConfigurationTestBase {
assertTrue(metricsConfiguration.isUptime());
assertTrue(metricsConfiguration.isLogging());
assertTrue(metricsConfiguration.isSecurityCaches());
+ assertTrue(metricsConfiguration.isExecutorServices());
}
private void verifyAddresses() {
diff --git
a/artemis-server/src/test/java/org/apache/activemq/artemis/uri/AcceptorParserTest.java
b/artemis-server/src/test/java/org/apache/activemq/artemis/uri/AcceptorParserTest.java
index 46eea43dfa..ecc20f0717 100644
---
a/artemis-server/src/test/java/org/apache/activemq/artemis/uri/AcceptorParserTest.java
+++
b/artemis-server/src/test/java/org/apache/activemq/artemis/uri/AcceptorParserTest.java
@@ -53,7 +53,7 @@ public class AcceptorParserTest {
assertEquals(33,
ConfigurationHelper.getIntProperty(TransportConstants.QUIET_PERIOD, -1,
configs.get(0).getParams()));
assertEquals(55,
ConfigurationHelper.getIntProperty(TransportConstants.SHUTDOWN_TIMEOUT, -1,
configs.get(0).getParams()));
- NettyAcceptor nettyAcceptor = new NettyAcceptor("name", null,
configs.get(0).getParams(), null, null, null, null, new HashMap<>());
+ NettyAcceptor nettyAcceptor = new NettyAcceptor("name", null,
configs.get(0).getParams(), null, null, null, null, new HashMap<>(),
"threadFactoryGroupName", null);
assertEquals(33, nettyAcceptor.getQuietPeriod());
assertEquals(55, nettyAcceptor.getShutdownTimeout());
diff --git
a/artemis-server/src/test/resources/ConfigurationTest-full-config.xml
b/artemis-server/src/test/resources/ConfigurationTest-full-config.xml
index f6ac7e8340..b9935e85bb 100644
--- a/artemis-server/src/test/resources/ConfigurationTest-full-config.xml
+++ b/artemis-server/src/test/resources/ConfigurationTest-full-config.xml
@@ -400,6 +400,7 @@
<uptime>true</uptime>
<logging>true</logging>
<security-caches>true</security-caches>
+ <executor-services>true</executor-services>
<plugin
class-name="org.apache.activemq.artemis.core.server.metrics.plugins.SimpleMetricsPlugin">
<property key="foo" value="x"/>
<property key="bar" value="y"/>
diff --git
a/artemis-server/src/test/resources/ConfigurationTest-xinclude-config.xml
b/artemis-server/src/test/resources/ConfigurationTest-xinclude-config.xml
index e6f3e827f0..803eed9747 100644
--- a/artemis-server/src/test/resources/ConfigurationTest-xinclude-config.xml
+++ b/artemis-server/src/test/resources/ConfigurationTest-xinclude-config.xml
@@ -288,6 +288,7 @@
<uptime>true</uptime>
<logging>true</logging>
<security-caches>true</security-caches>
+ <executor-services>true</executor-services>
<plugin
class-name="org.apache.activemq.artemis.core.server.metrics.plugins.SimpleMetricsPlugin">
<property key="foo" value="x"/>
<property key="bar" value="y"/>
diff --git
a/artemis-server/src/test/resources/ConfigurationTest-xinclude-schema-config-metrics.xml
b/artemis-server/src/test/resources/ConfigurationTest-xinclude-schema-config-metrics.xml
index 9d03ab63be..e5d4c8adff 100644
---
a/artemis-server/src/test/resources/ConfigurationTest-xinclude-schema-config-metrics.xml
+++
b/artemis-server/src/test/resources/ConfigurationTest-xinclude-schema-config-metrics.xml
@@ -24,6 +24,7 @@
<uptime>true</uptime>
<logging>true</logging>
<security-caches>true</security-caches>
+ <executor-services>true</executor-services>
<plugin
class-name="org.apache.activemq.artemis.core.server.metrics.plugins.SimpleMetricsPlugin">
<property key="foo" value="x"/>
<property key="bar" value="y"/>
diff --git a/artemis-server/src/test/resources/metrics.xml
b/artemis-server/src/test/resources/metrics.xml
index dd82a2f0dd..094050e17e 100644
--- a/artemis-server/src/test/resources/metrics.xml
+++ b/artemis-server/src/test/resources/metrics.xml
@@ -29,6 +29,7 @@
<uptime>true</uptime>
<logging>true</logging>
<security-caches>true</security-caches>
+ <executor-services>true</executor-services>
<plugin
class-name="org.apache.activemq.artemis.core.config.impl.FileConfigurationTest$FakeMetricPlugin">
<property key="key1" value="value1"/>
<property key="key2" value="value2"/>
diff --git
a/artemis-unit-test-support/src/main/java/org/apache/activemq/artemis/tests/extensions/ThreadLeakCheckDelegate.java
b/artemis-unit-test-support/src/main/java/org/apache/activemq/artemis/tests/extensions/ThreadLeakCheckDelegate.java
index 1a95258540..97c388b77c 100644
---
a/artemis-unit-test-support/src/main/java/org/apache/activemq/artemis/tests/extensions/ThreadLeakCheckDelegate.java
+++
b/artemis-unit-test-support/src/main/java/org/apache/activemq/artemis/tests/extensions/ThreadLeakCheckDelegate.java
@@ -239,7 +239,7 @@ public class ThreadLeakCheckDelegate {
return true;
} else if (threadName.contains("globalEventExecutor")) {
return true;
- } else if (threadName.contains("netty-threads")) {
+ } else if (threadName.contains("activemq-remoting") ||
threadName.contains("activemq-client-remoting")) {
// This is ok as we use EventLoopGroup.shutdownGracefully() which
will shutdown things with a bit of delay
// if the EventLoop's are still busy.
return true;
diff --git
a/artemis-web/src/main/java/org/apache/activemq/artemis/component/WebServerComponent.java
b/artemis-web/src/main/java/org/apache/activemq/artemis/component/WebServerComponent.java
index a3b07a6501..9ecc2dc6f3 100644
---
a/artemis-web/src/main/java/org/apache/activemq/artemis/component/WebServerComponent.java
+++
b/artemis-web/src/main/java/org/apache/activemq/artemis/component/WebServerComponent.java
@@ -26,6 +26,8 @@ import java.nio.file.Files;
import java.nio.file.LinkOption;
import java.nio.file.Path;
import java.nio.file.Paths;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.EnumSet;
@@ -33,6 +35,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.concurrent.ThreadFactory;
import java.util.stream.Collectors;
import jakarta.servlet.DispatcherType;
@@ -49,6 +52,7 @@ import org.apache.activemq.artemis.dto.ComponentDTO;
import org.apache.activemq.artemis.dto.WebServerDTO;
import org.apache.activemq.artemis.logs.AuditLogger;
import org.apache.activemq.artemis.marker.WebServerComponentMarker;
+import org.apache.activemq.artemis.utils.ActiveMQThreadFactory;
import org.apache.activemq.artemis.utils.ClassloadingUtil;
import org.apache.activemq.artemis.utils.PemConfigUtil;
import org.eclipse.jetty.alpn.server.ALPNServerConnectionFactory;
@@ -74,6 +78,8 @@ import org.eclipse.jetty.util.Scanner;
import org.eclipse.jetty.util.ssl.SslContextFactory;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.eclipse.jetty.util.thread.ScheduledExecutorScheduler;
+import org.eclipse.jetty.util.thread.Scheduler;
+import org.eclipse.jetty.util.thread.ThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -137,7 +143,10 @@ public class WebServerComponent implements
ExternalComponent, WebServerComponent
}
ActiveMQWebLogger.LOGGER.startingEmbeddedWebServer();
- server = new Server(new QueuedThreadPool(webServerConfig.maxThreads,
webServerConfig.minThreads, webServerConfig.idleThreadTimeout));
+ ThreadFactory threadFactory =
AccessController.doPrivileged((PrivilegedAction<ThreadFactory>) () -> new
ActiveMQThreadFactory("web", false, WebServerComponent.class.getClassLoader()));
+ ThreadPool threadPool = new QueuedThreadPool(webServerConfig.maxThreads,
webServerConfig.minThreads, webServerConfig.idleThreadTimeout, -1, null, null,
threadFactory);
+ Scheduler scheduler = new
ScheduledExecutorScheduler("activemq-web-scheduled", false);
+ server = new Server(threadPool, scheduler, null);
handlers = new Handler.Sequence();
HttpConfiguration httpConfiguration = new HttpConfiguration();
diff --git a/docs/user-manual/configuration-index.adoc
b/docs/user-manual/configuration-index.adoc
index 73e251cb89..4c2d941b9e 100644
--- a/docs/user-manual/configuration-index.adoc
+++ b/docs/user-manual/configuration-index.adoc
@@ -519,7 +519,7 @@ In most cases this should be set to '1'.
| <<resource-limit-type,a list of resource-limits>>
| n/a
-|
xref:thread-pooling.adoc#server-scheduled-thread-pool[scheduled-thread-pool-max-size]
+|
xref:thread-pooling.adoc#scheduled-thread-pool[scheduled-thread-pool-max-size]
| Maximum number of threads to use for the scheduled thread pool.
| 5
diff --git a/docs/user-manual/metrics.adoc b/docs/user-manual/metrics.adoc
index 4eee867854..19bd38f99d 100644
--- a/docs/user-manual/metrics.adoc
+++ b/docs/user-manual/metrics.adoc
@@ -104,58 +104,98 @@ However, these metrics can be deduced by aggregating the
lower level metrics (e.
There are a handful of other useful metrics that are related to the JVM, the
underlying operating system, etc.
These metrics are provided specifically by Micrometer and therefore do not
have the `artmemis.` prefix.
-JVM memory metrics::
+==== JVM memory metrics
Gauges buffer and memory pool utilization.
Underlying data gathered from Java's
https://docs.oracle.com/en/java/javase/11/docs/api/java.management/java/lang/management/BufferPoolMXBean.html[BufferPoolMXBeans]
and
https://docs.oracle.com/en/java/javase/11/docs/api/java.management/java/lang/management/MemoryPoolMXBean.html[MemoryPoolMXBeans].
-+
+
Enabled by default.
-JVM GC::
+
+==== JVM GC
Gauges max and live data size, promotion and allocation rates, and the number
of times the GC pauses (or concurrent phase time in the case of CMS).
Underlying data gathered from Java's
https://docs.oracle.com/en/java/javase/11/docs/api/java.management/java/lang/management/MemoryPoolMXBean.html[MemoryPoolMXBeans].
-+
+
Disabled by default.
-JVM thread::
+
+==== JVM thread
Gauges thread peak, the number of daemon threads, and live threads.
Underlying data gathered from Java's
https://docs.oracle.com/en/java/javase/11/docs/api/java.management/java/lang/management/ThreadMXBean.html[ThreadMXBean].
-+
+
Disabled by default.
-Netty::
+
+==== Netty Allocator
Collects metrics from Netty's
https://netty.io/4.1/api/io/netty/buffer/PooledByteBufAllocatorMetric.html[PooledByteBufAllocatorMetric].
-+
+
Disabled by default.
File descriptors::
Gauges current and max-allowed open files.
-+
+
Disabled by default.
-Processor::
+
+==== Processor
Gauges system CPU count, CPU usage, and 1-minute load average as well as
process CPU usage.
-+
+
Disabled by default.
-Uptime::
+
+==== Uptime
Gauges process start time and uptime.
-+
+
Disabled by default.
-Logging::
+
+==== Logging
Counts the number of logging events per logging category (e.g. `WARN`,
`ERROR`, etc.).
-+
+
Disabled by default.
-+
+
[WARNING]
====
This works _exclusively_ with Log4j2 (i.e the default logging implementation
shipped with the broker).
If you're embedding the broker and using a different logging implementation
(e.g. Log4j 1.x, JUL, Logback, etc.) and you enable these metrics then the
broker will fail to start with a `java.lang.NoClassDefFoundError` as it
attempts to locate Log4j2 classes that don't exist on the classpath.
====
-Security caches::
-The following authentication & authorization cache metrics are exported. They
are all tagged by `cache` (either `authentication` or `authorization`).
Additional tags are noted.
+
+==== Security caches
+The following authentication & authorization cache metrics are exported.
+They are all tagged by `cache` (either `authentication` or `authorization`).
+Additional tags are noted.
+
* `cache.size`
* `cache.puts`
* `cache.gets` tagged by `result` - either `hit` or `miss`
* `cache.evictions`
* `cache.eviction.weight`
-+
Disabled by default.
+==== Executor Services
+
+Metrics for executor services cover both the major instances of
https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/ExecutorService.html[`java.util.concurrent.ExecutorService`]
used by the broker to manage threads as well as
https://netty.io/4.1/api/io/netty/util/concurrent/EventExecutor.html[`EventExecutors`]
associated with instances of Netty's
https://netty.io/4.1/api/io/netty/channel/EventLoopGroup.html[`EventLoopGroup`].
+
+Executor service metrics are disabled by default.
+
+===== Java Executor Services
+
+All metrics are tagged with the name of the broker and with the executor
service name which corresponds to the role it plays within the broker (e.g.
general, io, paging, scheduled).
+
+* `executor`
+* `executor.completed`
+* `executor.active`
+* `executor.idle`
+* `executor.queued`
+* `executor.queue.remaining`
+* `executor.pool.core`
+* `executor.pool.size`
+* `executor.pool.max`
+
+Any `ExecutorService` instance used to schedule tasks also has these metrics:
+
+* `executor.scheduled.repetitively`
+* `executor.scheduled.once`
+
+====== Netty Event Executors
+
+All metrics are tagged with the name of the broker and with the name of the
underlying Netty `EventExecutor`.
+
+* `netty.eventexecutor.tasks.pending`
+
== Configuration
Metrics for all addresses and queues are enabled by default.
@@ -176,6 +216,7 @@ Here's a configuration with all optional metrics:
<uptime>true</uptime> <!-- defaults to false -->
<logging>true</logging> <!-- defaults to false -->
<security-caches>true</security-caches> <!-- defaults to false -->
+ <executor-services>true</executor-services> <!-- defaults to false -->
<plugin
class-name="org.apache.activemq.artemis.core.server.metrics.plugins.LoggingMetricsPlugin"/>
</metrics>
----
diff --git a/docs/user-manual/thread-pooling.adoc
b/docs/user-manual/thread-pooling.adoc
index 5cd227a8a6..fd99397e45 100644
--- a/docs/user-manual/thread-pooling.adoc
+++ b/docs/user-manual/thread-pooling.adoc
@@ -5,62 +5,111 @@
This chapter describes how Apache ActiveMQ Artemis uses and pools threads and
how you can manage them.
-First we'll discuss how threads are managed and used on the server side, then
we'll look at the client side.
+First we'll discuss how threads are managed and used on the server side then
we'll look at the client side.
== Server-Side Thread Management
-Each Apache ActiveMQ Artemis Server maintains a single thread pool for general
use, and a scheduled thread pool for scheduled use.
-A Java scheduled thread pool cannot be configured to use a standard thread
pool, otherwise we could use a single thread pool for both scheduled and non
scheduled activity.
+Thread pools exist for each of the following:
-Apache ActiveMQ Artemis will, by default, cap its thread pool at three times
the number of cores (or hyper-threads) as reported by `
Runtime.getRuntime().availableProcessors()`` for processing incoming packets.
-To override this value, you can set the number of threads by specifying the
parameter ``nioRemotingThreads` in the transport configuration.
-See the xref:configuring-transports.adoc#configuring-the-transport[configuring
transports] for more information on this.
+* scheduled tasks
+* general use
+* asynchronous IO
+* paging
+* remoting (managed by Netty on a per-acceptor basis)
-There are also a small number of other places where threads are used directly,
we'll discuss each in turn.
+[IMPORTANT]
+.Broker Identification in Thread Names
+====
+Many thread names contain broker identification.
+This is done to assist in cases where multiple brokers are running in the same
JVM (e.g. in the test-suite).
+The identification which appears in the thread name is determined by the
following in order of precedence:
-=== Server Scheduled Thread Pool
+* `identity` set in the internal `ConfigurationImpl` object (done by tests)
+* `name` set in `broker.xml`
+* the hexadecimal representation of the `ActiveMQServerImpl` Java Object's
identity acquired via
https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/lang/System.html#identityHashCode(java.lang.Object)[`System.identityHashCode`]
+====
-The server scheduled thread pool is used for most activities on the server
side that require running periodically or with delays.
-It maps internally to a `java.util.concurrent.ScheduledThreadPoolExecutor`
instance.
+=== Scheduled Thread Pool
-The maximum number of thread used by this pool is configure in `broker.xml`
with the `scheduled-thread-pool-max-size` parameter.
-The default value is `5` threads.
-A small number of threads is usually sufficient for this pool.
+The scheduled thread pool is used for most activities on the server side that
require running periodically or with delays.
+This includes tasks like scanning:
-=== General Purpose Server Thread Pool
+* queues for expired messages or scheduled deliveries
+* configuration files for changes
+* unused addresses & queues for deletion
-This general purpose thread pool is used for most asynchronous actions on the
server side.
-It maps internally to a `java.util.concurrent.ThreadPoolExecutor` instance.
+The maximum number of thread used by this pool is configure in `broker.xml`
with the `scheduled-thread-pool-max-size` parameter, e.g.:
+
+[,xml]
+----
+<scheduled-thread-pool-max-size>10</scheduled-thread-pool-max-size>
+----
-The maximum number of thread used by this pool is configure in `broker.xml`
with the `thread-pool-max-size` parameter.
+The default `scheduled-thread-pool-max-size` is `5` . A value of `0` is not
allowed.
-If a value of `-1` is used this signifies that the thread pool has no upper
bound and new threads will be created on demand if there are not enough threads
available to satisfy a request.
-If activity later subsides then threads are timed-out and closed.
+The name for threads from this pool will contain `activemq-scheduled`.
-If a value of `n` where ``n``is a positive integer greater than zero is used
this signifies that the thread pool is bounded.
-If more requests come in and there are no free threads in the pool and the
pool is full then requests will block until a thread becomes available.
-It is recommended that a bounded thread pool is used with caution since it can
lead to dead-lock situations if the upper bound is chosen to be too low.
+=== General Purpose Thread Pool
-The default value for `thread-pool-max-size` is `30`.
+This general purpose thread pool is used for most asynchronous actions on the
server side.
+The maximum number of threads used by this pool is configure in `broker.xml`
with the `thread-pool-max-size` parameter, e.g.:
-See the
https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ThreadPoolExecutor.html[J2SE
javadoc] for more information on unbounded (cached), and bounded (fixed)
thread pools.
+[,xml]
+----
+<thread-pool-max-size>60</thread-pool-max-size>
+----
-=== Expiry Reaper Thread
+The default `thread-pool-max-size` is `30`.
+A value of `-1` signifies that the thread pool has _no upper bound_ and new
threads will be created on demand if there are not enough threads already
available to satisfy demand.
+A value of `0` is not allowed.
-A single thread is also used on the server side to scan for expired messages
in queues.
-We cannot use either of the thread pools for this since this thread needs to
run at its own configurable priority.
+Any threads in this pool which are idle for `60` seconds will be terminated.
-For more information on configuring the reaper, please see
xref:message-expiry.adoc#message-expiry[message expiry].
+The name for threads from this pool will contain `activemq-<brokerName>`.
=== Asynchronous IO
-Asynchronous IO has a thread pool for receiving and dispatching events out of
the native layer.
-You will find it on a thread dump with the prefix ActiveMQ-AIO-poller-pool.
-Apache ActiveMQ Artemis uses one thread per opened file on the journal (there
is usually one).
+Threads from this pool are used for journal-related disk I/O, JDBC, &
replication.
+
+The name for threads from this pool will contain `activemq-io-<brokerName>`.
+
+=== Paging
+
+Threads from this pool are used to write to and read from paging (disk or
JDBC).
+
+The name for threads from this pool will contain
`activemq-paging-<brokerName>`.
+
+=== Netty Acceptors
+
+Apache ActiveMQ Artemis will, by default, cap Netty threads on a per-acceptor
basis at three times the number of cores (or hyper-threads) as reported by
`Runtime.getRuntime().availableProcessors()` for processing network traffic.
+To override this value, you can set the number of threads by specifying the
parameter `remotingThreads` in the transport configuration.
+See the xref:configuring-transports.adoc#configuring-the-transport[configuring
transports] for more information on this.
+
+Threads names will include the name of their corresponding acceptor with the
prefix `activemq-remoting-`.
+For example, for the acceptor named `amqp` the corresponding thread names will
contain `activemq-remoting-amqp-<brokerName>`.
+
+=== Other Server-Side Threads
+
+A thread dump from the server's JVM will have other threads as well.
+Here's other names you might find in a thread dump and the function they
perform.
+
+`activemq-web`::
+thread pool managed by Jetty (i.e. the xref:web-server.adoc[embedded web
server]) to handle HTTP connections (e.g. from the web console or other Jolokia
clients)
+`activemq-failure-check-thread`::
+checks TTL on incoming connections
+`activemq-buffer-timeout`::
+flushes disk IO buffers upon timeout
+`activemq-libaio-poller`::
+polls AIO for callbacks
+`activemq-critical-analyzer`::
+monitors for timeouts of various critical server operations
+`activemq-shutdown-timer`::
+monitors configuration directory for status file to stop the server
+`activemq-remoting-service`::
+in-vm connectivity and invoking failure listeners for Netty
+`Log4j2-TF-\*-Scheduled-*`::
+executes Log4j2 tasks related to `CronTriggeringPolicy` used by default
`log4j2.properties`
-There is also a single thread used to invoke writes on libaio.
-We do that to avoid context switching on libaio that would cause performance
issues.
-You will find this thread on a thread dump with the prefix
ActiveMQ-AIO-writer-pool.
== Client-Side Thread Management
diff --git a/docs/user-manual/versions.adoc b/docs/user-manual/versions.adoc
index 2639a6bd8b..4fed10de2c 100644
--- a/docs/user-manual/versions.adoc
+++ b/docs/user-manual/versions.adoc
@@ -13,6 +13,59 @@ NOTE: If the upgrade spans multiple versions then the steps
from *each* version
NOTE: Follow the general upgrade procedure outlined in the
xref:upgrading.adoc#upgrading-the-broker[Upgrading the Broker] chapter in
addition to any version-specific upgrade instructions outlined here.
+== 2.43.0
+
+https://issues.apache.org/jira/secure/ReleaseNote.jspa?projectId=12315920&version=XXX[Full
release notes]
+
+=== Highlights
+
+* Broker observability has improved with the new ability to export
xref:metrics.adoc#executor-services[metrics for executor services] (aka thread
pools).
+
+=== Upgrading from 2.41.0
+
+* As part of the work for
https://issues.apache.org/jira/browse/ARTEMIS-5557[ARTEMIS-5557] to report
executor service metrics the names of many ActiveMQ-related threads were
changed to be more clear and consistent.
+This will be visible, for example, when inspecting thread dumps or if you
happen to include the thread name in your logging output.
+Here's examples of old names vs. the new ones.
++
+|===
+|Old Thread Name |New Thread Name
+
+|Thread-0 (ActiveMQ-scheduled-threads)
+|Thread-0 (activemq-scheduled-0.0.0.0)
+
+|Thread-0 (ActiveMQ-server-ActiveMQServerImpl::name=0.0.0.0)
+|Thread-0 (activemq-0.0.0.0)
+
+|Thread-0 (ActiveMQ-IO-server-ActiveMQServerImpl::name=0.0.0.0)
+|Thread-0 (activemq-io-0.0.0.0)
+
+|Thread-0
(ActiveMQ-PageExecutor-server-org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl$2@7841111)
+|Thread-0 (activemq-paging-0.0.0.0)
+
+|Thread-0 (activemq-netty-threads)
+|Thread-0 (activemq-remoting-amqp-0.0.0.0)
+
+|ActiveMQ Artemis Server Shutdown Timer
+|activemq-shutdown-timer
+
+|Critical-Analyzer-0 (CriticalAnalyzer)
+|Thread-0 (activemq-critical-analyzer)
+
+|Network-Checker-0 (NetworkChecker)
+|Thread-0 (activemq-network-checker)
+
+|Apache ActiveMQ Artemis libaio poller
+|activemq-libaio-poller
+
+|qtp748316439-74
+|Thread-0 (activemq-web)
+
+|Scheduler-624545052-1
+|activemq-web-scheduled-1
+|===
++
+See the xref:thread-pooling.adoc[Thread Pooling] documentation for more
details.
+
== 2.42.0
https://issues.apache.org/jira/secure/ReleaseNote.jspa?projectId=12315920&version=12355900[Full
release notes]
diff --git
a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/largemessage/ServerLargeMessageTest.java
b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/largemessage/ServerLargeMessageTest.java
index 9f7e309c7c..6363f7281b 100644
---
a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/largemessage/ServerLargeMessageTest.java
+++
b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/largemessage/ServerLargeMessageTest.java
@@ -377,7 +377,7 @@ public class ServerLargeMessageTest extends
ActiveMQTestBase {
private SequentialFile originalFile;
MockSequentialFile(SequentialFile originalFile) throws Exception {
- super(originalFile.getJavaFile().getParentFile(),
originalFile.getFileName(), new FakeSequentialFileFactory(), null);
+ super(originalFile.getJavaFile().getParentFile(),
originalFile.getFileName(), new FakeSequentialFileFactory());
this.originalFile = originalFile;
this.originalFile.close();
}
diff --git
a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/plugin/ExecutorServiceMetricsTest.java
b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/plugin/ExecutorServiceMetricsTest.java
new file mode 100644
index 0000000000..9f8c147790
--- /dev/null
+++
b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/plugin/ExecutorServiceMetricsTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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.activemq.artemis.tests.integration.plugin;
+
+import java.util.Arrays;
+import java.util.Set;
+
+import io.micrometer.core.instrument.Meter;
+import io.micrometer.core.instrument.Tag;
+import io.micrometer.core.instrument.Tags;
+import org.apache.activemq.artemis.core.config.MetricsConfiguration;
+import org.apache.activemq.artemis.core.remoting.impl.netty.NettyAcceptor;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.metrics.BrokerMetricNames;
+import
org.apache.activemq.artemis.core.server.metrics.plugins.SimpleMetricsPlugin;
+import org.apache.activemq.artemis.spi.core.remoting.Acceptor;
+import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class ExecutorServiceMetricsTest extends ActiveMQTestBase {
+
+ @Test
+ public void testExecutorServiceMetricsPositive() throws Exception {
+ internalTestMetrics(true);
+ }
+
+ @Test
+ public void testExecutorServiceMetricsNegative() throws Exception {
+ internalTestMetrics(false);
+ }
+
+ private void internalTestMetrics(boolean enabled) throws Exception {
+ ActiveMQServer server = createServer(false,
createDefaultNettyConfig().addAcceptorConfiguration("foo",
"tcp://127.0.0.1:61617").setMetricsConfiguration(new
MetricsConfiguration().setPlugin(new
SimpleMetricsPlugin().init(null)).setExecutorServices(enabled)));
+ server.start();
+ Set<Meter.Id> metrics = MetricsPluginTest.getMetrics(server).keySet();
+ for (String nameTagValue :
Arrays.asList(BrokerMetricNames.GENERAL_EXECUTOR_SERVICE,
BrokerMetricNames.IO_EXECUTOR_SERVICE, BrokerMetricNames.PAGE_EXECUTOR_SERVICE,
BrokerMetricNames.SCHEDULED_EXECUTOR_SERVICE)) {
+ Tags defaultTags = Tags.of(Tag.of("broker",
server.getConfiguration().getName()), Tag.of("name", nameTagValue));
+ assertEquals(enabled, metrics.contains(new Meter.Id("executor",
defaultTags, null, null, null)));
+ assertEquals(enabled, metrics.contains(new
Meter.Id("executor.completed", defaultTags, null, null, null)));
+ assertEquals(enabled, metrics.contains(new
Meter.Id("executor.active", defaultTags, null, null, null)));
+ assertEquals(enabled, metrics.contains(new Meter.Id("executor.idle",
defaultTags, null, null, null)));
+ assertEquals(enabled, metrics.contains(new
Meter.Id("executor.queued", defaultTags, null, null, null)));
+ assertEquals(enabled, metrics.contains(new
Meter.Id("executor.queue.remaining", defaultTags, null, null, null)));
+ assertEquals(enabled, metrics.contains(new
Meter.Id("executor.pool.core", defaultTags, null, null, null)));
+ assertEquals(enabled, metrics.contains(new
Meter.Id("executor.pool.size", defaultTags, null, null, null)));
+ assertEquals(enabled, metrics.contains(new
Meter.Id("executor.pool.max", defaultTags, null, null, null)));
+ if
(nameTagValue.equals(BrokerMetricNames.SCHEDULED_EXECUTOR_SERVICE)) {
+ assertEquals(enabled, metrics.contains(new
Meter.Id("executor.scheduled.repetitively", defaultTags, null, null, null)));
+ assertEquals(enabled, metrics.contains(new
Meter.Id("executor.scheduled.once", defaultTags, null, null, null)));
+ }
+ }
+ boolean nettyAcceptorExists = false;
+ for (Acceptor acceptor :
server.getRemotingService().getAcceptors().values()) {
+ if (acceptor instanceof NettyAcceptor nettyAcceptor) {
+ nettyAcceptorExists = true;
+ for (int i = 0; i < nettyAcceptor.getRemotingThreads(); i++) {
+ Tags defaultTags = Tags.of(Tag.of("broker", "localhost"),
Tag.of("name", "Thread-" + i + " (activemq-remoting-" + nettyAcceptor.getName()
+ "-" + server.getConfiguration().getName() + ")"));
+ assertEquals(enabled, metrics.contains(new
Meter.Id("netty.eventexecutor.tasks.pending", defaultTags, null, null, null)));
+ }
+ acceptor.stop();
+ metrics = MetricsPluginTest.getMetrics(server).keySet();
+ for (int i = 0; i < nettyAcceptor.getRemotingThreads(); i++) {
+ Tags defaultTags = Tags.of(Tag.of("broker", "localhost"),
Tag.of("name", "Thread-" + i + " (activemq-remoting-" + nettyAcceptor.getName()
+ "-" + server.getConfiguration().getName() + ")"));
+ assertFalse(metrics.contains(new
Meter.Id("netty.eventexecutor.tasks.pending", defaultTags, null, null, null)));
+ }
+ }
+ }
+ assertTrue(nettyAcceptorExists);
+ }
+}
\ No newline at end of file
diff --git
a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/SharedNothingReplicationFlowControlTest.java
b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/SharedNothingReplicationFlowControlTest.java
index 44e8032270..2b1b62d83f 100644
---
a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/SharedNothingReplicationFlowControlTest.java
+++
b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/SharedNothingReplicationFlowControlTest.java
@@ -16,21 +16,17 @@
*/
package org.apache.activemq.artemis.tests.integration.replication;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
-
import java.io.File;
import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
+import java.lang.invoke.MethodHandles;
import java.lang.management.ManagementFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@@ -76,12 +72,12 @@ import
org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.ActiveMQServers;
import org.apache.activemq.artemis.core.server.JournalType;
import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
-import org.apache.activemq.artemis.tests.util.Wait;
import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
import
org.apache.activemq.artemis.spi.core.security.ActiveMQJAASSecurityManager;
import org.apache.activemq.artemis.spi.core.security.jaas.InVMLoginModule;
import org.apache.activemq.artemis.tests.extensions.TargetTempDirFactory;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.apache.activemq.artemis.tests.util.Wait;
import org.apache.activemq.artemis.utils.ExecutorFactory;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -89,7 +85,10 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.lang.invoke.MethodHandles;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
public class SharedNothingReplicationFlowControlTest extends ActiveMQTestBase {
@@ -324,7 +323,7 @@ public class SharedNothingReplicationFlowControlTest
extends ActiveMQTestBase {
@Override
public SequentialFile createSequentialFile(String fileName) {
- return new TestableSequentialFile(this, journalDir, fileName, maxIO,
writeExecutor);
+ return new TestableSequentialFile(this, journalDir, fileName, maxIO);
}
private static File newFolder(File root, String... subDirs) throws
IOException {
@@ -360,9 +359,8 @@ public class SharedNothingReplicationFlowControlTest
extends ActiveMQTestBase {
public TestableSequentialFile(SequentialFileFactory factory,
File directory,
String file,
- int maxIO,
- Executor writerExecutor) {
- super(factory, directory, file, maxIO, writerExecutor);
+ int maxIO) {
+ super(factory, directory, file, maxIO);
}
@Override
diff --git
a/tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/nettynative/NettyNativeTest.java
b/tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/nettynative/NettyNativeTest.java
index 849370b4a0..7b83e34bc4 100644
---
a/tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/nettynative/NettyNativeTest.java
+++
b/tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/nettynative/NettyNativeTest.java
@@ -37,6 +37,7 @@ import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import java.io.File;
+import java.util.Arrays;
public class NettyNativeTest extends SmokeTestBase {
@@ -92,8 +93,10 @@ public class NettyNativeTest extends SmokeTestBase {
}
File artemisLog = new File("target/" + SERVER_NAME + "/log/artemis.log");
- assertTrue(findLogRecord(artemisLog, "Acceptor using native"));
- assertFalse(findLogRecord(artemisLog, "Acceptor using nio"));
+ for (String protocol : Arrays.asList("artemis", "amqp", "stomp",
"hornetq", "mqtt")) {
+ assertTrue(findLogRecord(artemisLog, "Acceptor " + protocol + " using
native"));
+ assertFalse(findLogRecord(artemisLog, "Acceptor " + protocol + "
using nio"));
+ }
}
}
diff --git
a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/journal/impl/TimedBufferTest.java
b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/journal/impl/TimedBufferTest.java
index bd83d823f8..a7a6f7755e 100644
---
a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/journal/impl/TimedBufferTest.java
+++
b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/journal/impl/TimedBufferTest.java
@@ -227,7 +227,7 @@ public class TimedBufferTest extends ActiveMQTestBase {
NIOSequentialFileFactory factory = new
NIOSequentialFileFactory(getTestDirfile(), 1) {
@Override
public SequentialFile createSequentialFile(final String fileName) {
- return new NIOSequentialFile(this, journalDir, fileName, maxIO,
writeExecutor) {
+ return new NIOSequentialFile(this, journalDir, fileName, maxIO) {
@Override
protected void syncChannel(FileChannel channel) throws
IOException {
try {
diff --git
a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/impl/netty/NettyAcceptorFactoryTest.java
b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/impl/netty/NettyAcceptorFactoryTest.java
index 351e7d1eb3..7d297f4b3b 100644
---
a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/impl/netty/NettyAcceptorFactoryTest.java
+++
b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/impl/netty/NettyAcceptorFactoryTest.java
@@ -68,7 +68,7 @@ public class NettyAcceptorFactoryTest extends
ActiveMQTestBase {
};
- Acceptor acceptor = factory.createAcceptor("netty", null, params,
handler, listener,
Executors.newCachedThreadPool(ActiveMQThreadFactory.defaultThreadFactory(getClass().getName())),
Executors.newScheduledThreadPool(ActiveMQDefaultConfiguration.getDefaultScheduledThreadPoolMaxSize(),
ActiveMQThreadFactory.defaultThreadFactory(getClass().getName())), new
HashMap<>());
+ Acceptor acceptor = factory.createAcceptor("netty", null, params,
handler, listener,
Executors.newCachedThreadPool(ActiveMQThreadFactory.defaultThreadFactory(getClass().getName())),
Executors.newScheduledThreadPool(ActiveMQDefaultConfiguration.getDefaultScheduledThreadPoolMaxSize(),
ActiveMQThreadFactory.defaultThreadFactory(getClass().getName())), new
HashMap<>(), null, null);
assertInstanceOf(NettyAcceptor.class, acceptor);
}
diff --git
a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/impl/netty/NettyAcceptorTest.java
b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/impl/netty/NettyAcceptorTest.java
index 4e4b99f6d5..3f63901fde 100644
---
a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/impl/netty/NettyAcceptorTest.java
+++
b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/impl/netty/NettyAcceptorTest.java
@@ -96,7 +96,7 @@ public class NettyAcceptorTest extends ActiveMQTestBase {
};
pool2 =
Executors.newScheduledThreadPool(ActiveMQDefaultConfiguration.getDefaultScheduledThreadPoolMaxSize(),
ActiveMQThreadFactory.defaultThreadFactory(getClass().getName()));
pool3 =
Executors.newSingleThreadExecutor(ActiveMQThreadFactory.defaultThreadFactory(getClass().getName()));
- NettyAcceptor acceptor = new NettyAcceptor("netty", null, params,
handler, listener, pool2, pool3, new HashMap<>());
+ NettyAcceptor acceptor = new NettyAcceptor("netty", null, params,
handler, listener, pool2, pool3, new HashMap<>(), null, null);
addActiveMQComponent(acceptor);
acceptor.start();
@@ -149,7 +149,7 @@ public class NettyAcceptorTest extends ActiveMQTestBase {
params.put(TransportConstants.SSL_ENABLED_PROP_NAME, "true");
try {
- new NettyAcceptor("netty", null, params, null, null, null, null,
Map.of());
+ new NettyAcceptor("netty", null, params, null, null, null, null,
Map.of(), null, null);
fail("This should have failed with an IllegalArgumentException");
} catch (IllegalArgumentException e) {
// expected
@@ -161,7 +161,7 @@ public class NettyAcceptorTest extends ActiveMQTestBase {
Map<String, Object> params = new HashMap<>();
params.put(TransportConstants.SSL_ENABLED_PROP_NAME, "true");
params.put(TransportConstants.KEYSTORE_PROVIDER_PROP_NAME,
RandomUtil.randomUUIDString());
- new NettyAcceptor("netty", null, params, null, null, null, null,
Map.of());
+ new NettyAcceptor("netty", null, params, null, null, null, null,
Map.of(), null, null);
}
@Test
@@ -169,6 +169,6 @@ public class NettyAcceptorTest extends ActiveMQTestBase {
Map<String, Object> params = new HashMap<>();
params.put(TransportConstants.SSL_ENABLED_PROP_NAME, "true");
params.put(TransportConstants.SSL_CONTEXT_PROP_NAME,
RandomUtil.randomUUIDString());
- new NettyAcceptor("netty", null, params, null, null, null, null,
Map.of());
+ new NettyAcceptor("netty", null, params, null, null, null, null,
Map.of(), null, null);
}
}
diff --git
a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/server/impl/fake/FakeAcceptorFactory.java
b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/server/impl/fake/FakeAcceptorFactory.java
index 07a7a50c00..7ccb25289b 100644
---
a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/server/impl/fake/FakeAcceptorFactory.java
+++
b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/server/impl/fake/FakeAcceptorFactory.java
@@ -25,6 +25,7 @@ import org.apache.activemq.artemis.api.core.BaseInterceptor;
import org.apache.activemq.artemis.core.security.ActiveMQPrincipal;
import org.apache.activemq.artemis.core.server.cluster.ClusterConnection;
import org.apache.activemq.artemis.core.server.management.NotificationService;
+import org.apache.activemq.artemis.core.server.metrics.MetricsManager;
import org.apache.activemq.artemis.spi.core.protocol.ProtocolManager;
import org.apache.activemq.artemis.spi.core.remoting.Acceptor;
import org.apache.activemq.artemis.spi.core.remoting.AcceptorFactory;
@@ -43,7 +44,9 @@ public class FakeAcceptorFactory implements AcceptorFactory {
ServerConnectionLifeCycleListener listener,
Executor threadPool,
ScheduledExecutorService scheduledThreadPool,
- Map<String, ProtocolManager> protocolMap) {
+ Map<String, ProtocolManager> protocolMap,
+ String threadFactoryGroupName,
+ MetricsManager metricsManager) {
return new FakeAcceptor();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
For further information, visit: https://activemq.apache.org/contact
