You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ex...@apache.org on 2023/05/25 20:03:28 UTC
[nifi] branch support/nifi-1.x updated: NIFI-11580 Use daemon threads in DistributedCacheClient for clean shutdown
This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch support/nifi-1.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/support/nifi-1.x by this push:
new f0b35dfa44 NIFI-11580 Use daemon threads in DistributedCacheClient for clean shutdown
f0b35dfa44 is described below
commit f0b35dfa4485281e81ba08719f96c50466c09a82
Author: Mike Moser <mo...@apache.org>
AuthorDate: Mon May 22 19:45:50 2023 +0000
NIFI-11580 Use daemon threads in DistributedCacheClient for clean shutdown
- Added thread naming based on Component Type and Component Identifier
This closes #7281
Signed-off-by: David Handermann <ex...@apache.org>
(cherry picked from commit a446af8aaaa7b65dd216043c847cfcb599bc83e8)
---
.../distributed/cache/client/CacheClientChannelPoolFactory.java | 8 ++++++--
.../nifi/distributed/cache/client/DistributedCacheClient.java | 7 +++++--
.../cache/client/DistributedMapCacheClientService.java | 1 +
.../cache/client/DistributedSetCacheClientService.java | 3 ++-
.../distributed/cache/client/NettyDistributedMapCacheClient.java | 4 +++-
.../distributed/cache/client/NettyDistributedSetCacheClient.java | 7 +++++--
6 files changed, 22 insertions(+), 8 deletions(-)
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/CacheClientChannelPoolFactory.java b/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/CacheClientChannelPoolFactory.java
index 7499c68d52..82a529ecd9 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/CacheClientChannelPoolFactory.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/CacheClientChannelPoolFactory.java
@@ -26,6 +26,7 @@ import io.netty.channel.pool.ChannelPool;
import io.netty.channel.pool.ChannelPoolHandler;
import io.netty.channel.pool.FixedChannelPool;
import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.util.concurrent.DefaultThreadFactory;
import org.apache.nifi.event.transport.netty.channel.pool.InitializingChannelPoolHandler;
import org.apache.nifi.remote.VersionNegotiatorFactory;
import org.apache.nifi.ssl.SSLContextService;
@@ -41,6 +42,7 @@ import java.time.Duration;
public class CacheClientChannelPoolFactory {
private static final int MAX_PENDING_ACQUIRES = 1024;
+ private static final boolean DAEMON_THREAD_ENABLED = true;
private int maxConnections = Runtime.getRuntime().availableProcessors() * 2;
@@ -62,15 +64,17 @@ public class CacheClientChannelPoolFactory {
* @param sslContextService the SSL context (if any) associated with requests to the service; if not specified,
* communications will not be encrypted
* @param factory creator of object used to broker the version of the distributed cache protocol with the service
+ * @param poolName channel pool name, used for threads name prefix
* @return a channel pool object from which {@link Channel} objects may be obtained
*/
public ChannelPool createChannelPool(final String hostname,
final int port,
final int timeoutMillis,
final SSLContextService sslContextService,
- final VersionNegotiatorFactory factory) {
+ final VersionNegotiatorFactory factory,
+ final String poolName) {
final SSLContext sslContext = (sslContextService == null) ? null : sslContextService.createContext();
- final EventLoopGroup group = new NioEventLoopGroup();
+ final EventLoopGroup group = new NioEventLoopGroup(new DefaultThreadFactory(poolName, DAEMON_THREAD_ENABLED));
final Bootstrap bootstrap = new Bootstrap();
final CacheClientChannelInitializer initializer = new CacheClientChannelInitializer(sslContext, factory, Duration.ofMillis(timeoutMillis), Duration.ofMillis(timeoutMillis));
bootstrap.group(group)
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedCacheClient.java b/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedCacheClient.java
index b2bfac78e8..8ebf9bacbd 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedCacheClient.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedCacheClient.java
@@ -45,14 +45,17 @@ public class DistributedCacheClient {
* @param sslContextService the SSL context (if any) associated with requests to the service; if not specified,
* communications will not be encrypted
* @param factory creator of object used to broker the version of the distributed cache protocol with the service
+ * @param identifier uniquely identifies this client
*/
protected DistributedCacheClient(final String hostname,
final int port,
final int timeoutMillis,
final SSLContextService sslContextService,
- final VersionNegotiatorFactory factory) {
+ final VersionNegotiatorFactory factory,
+ final String identifier) {
+ String poolName = String.format("%s[%s]", getClass().getSimpleName(), identifier);
this.channelPool = new CacheClientChannelPoolFactory().createChannelPool(
- hostname, port, timeoutMillis, sslContextService, factory);
+ hostname, port, timeoutMillis, sslContextService, factory, poolName);
}
/**
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedMapCacheClientService.java b/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedMapCacheClientService.java
index d46c79feb4..775b3932e4 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedMapCacheClientService.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedMapCacheClientService.java
@@ -114,6 +114,7 @@ public class DistributedMapCacheClientService extends AbstractControllerService
context.getProperty(COMMUNICATIONS_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue(),
context.getProperty(SSL_CONTEXT_SERVICE).asControllerService(SSLContextService.class),
versionNegotiatorFactory,
+ this.getIdentifier(),
getLogger());
}
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedSetCacheClientService.java b/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedSetCacheClientService.java
index 88b00c659a..19392ba51a 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedSetCacheClientService.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedSetCacheClientService.java
@@ -100,7 +100,8 @@ public class DistributedSetCacheClientService extends AbstractControllerService
context.getProperty(PORT).asInteger(),
context.getProperty(COMMUNICATIONS_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue(),
context.getProperty(SSL_CONTEXT_SERVICE).asControllerService(SSLContextService.class),
- versionNegotiatorFactory);
+ versionNegotiatorFactory,
+ this.getIdentifier());
}
@OnDisabled
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/NettyDistributedMapCacheClient.java b/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/NettyDistributedMapCacheClient.java
index 5be429c1b3..f87a5cf623 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/NettyDistributedMapCacheClient.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/NettyDistributedMapCacheClient.java
@@ -53,6 +53,7 @@ public class NettyDistributedMapCacheClient extends DistributedCacheClient {
* @param sslContextService the SSL context (if any) associated with requests to the service; if not specified,
* communications will not be encrypted
* @param factory creator of object used to broker the version of the distributed cache protocol with the service
+ * @param identifier uniquely identifies this client
* @param log Component Log from instantiating Services
*/
public NettyDistributedMapCacheClient(
@@ -61,9 +62,10 @@ public class NettyDistributedMapCacheClient extends DistributedCacheClient {
final int timeoutMillis,
final SSLContextService sslContextService,
final VersionNegotiatorFactory factory,
+ final String identifier,
final ComponentLog log
) {
- super(hostname, port, timeoutMillis, sslContextService, factory);
+ super(hostname, port, timeoutMillis, sslContextService, factory, identifier);
this.log = Objects.requireNonNull(log, "Component Log required");
}
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/NettyDistributedSetCacheClient.java b/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/NettyDistributedSetCacheClient.java
index 1734f87ef3..9bcc64a6d4 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/NettyDistributedSetCacheClient.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/NettyDistributedSetCacheClient.java
@@ -40,14 +40,17 @@ public class NettyDistributedSetCacheClient extends DistributedCacheClient {
* @param sslContextService the SSL context (if any) associated with requests to the service; if not specified,
* communications will not be encrypted
* @param factory creator of object used to broker the version of the distributed cache protocol with the service
+ * @param identifier uniquely identifies this client
*/
public NettyDistributedSetCacheClient(
final String hostname,
final int port,
final int timeoutMillis,
final SSLContextService sslContextService,
- final VersionNegotiatorFactory factory) {
- super(hostname, port, timeoutMillis, sslContextService, factory);
+ final VersionNegotiatorFactory factory,
+ final String identifier
+ ) {
+ super(hostname, port, timeoutMillis, sslContextService, factory, identifier);
}
/**