You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/04/24 16:12:00 UTC
(pulsar) branch master updated: [fix][broker] Fix broker shutdown delay by resolving hanging health checks (#24210)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 12961caf496 [fix][broker] Fix broker shutdown delay by resolving hanging health checks (#24210)
12961caf496 is described below
commit 12961caf4967b03634bb443d4718eb30a2a1f0ae
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Apr 24 19:11:51 2025 +0300
[fix][broker] Fix broker shutdown delay by resolving hanging health checks (#24210)
---
.../org/apache/pulsar/broker/PulsarService.java | 319 +++++++++++---------
.../pulsar/broker/admin/impl/BrokersBase.java | 169 +----------
.../pulsar/broker/service/BrokerService.java | 6 +-
.../pulsar/broker/service/HealthChecker.java | 330 +++++++++++++++++++++
.../broker/admin/AdminApiHealthCheckTest.java | 23 +-
.../extensions/BrokerRegistryIntegrationTest.java | 2 +-
...BrokerRegistryMetadataStoreIntegrationTest.java | 2 +-
.../systopic/PartitionedSystemTopicTest.java | 10 +-
.../testcontext/NonStartableTestPulsarService.java | 6 +-
9 files changed, 540 insertions(+), 327 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index c63f17c031e..51eecb53f9d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -21,7 +21,6 @@ package org.apache.pulsar.broker;
import static com.google.common.base.Preconditions.checkNotNull;
import static org.apache.commons.lang3.StringUtils.isBlank;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
-import static org.apache.pulsar.broker.admin.impl.BrokersBase.getHeartbeatTopicName;
import static org.apache.pulsar.broker.resourcegroup.ResourceUsageTransportManager.DISABLE_RESOURCE_USAGE_TRANSPORT_MANAGER;
import static org.apache.pulsar.common.naming.SystemTopicNames.isTransactionInternalName;
import com.google.common.annotations.VisibleForTesting;
@@ -75,7 +74,6 @@ import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.mledger.LedgerOffloader;
import org.apache.bookkeeper.mledger.LedgerOffloaderFactory;
import org.apache.bookkeeper.mledger.LedgerOffloaderStats;
-import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
import org.apache.bookkeeper.mledger.impl.NullLedgerOffloader;
import org.apache.bookkeeper.mledger.offload.Offloaders;
@@ -105,6 +103,7 @@ import org.apache.pulsar.broker.resources.ClusterResources;
import org.apache.pulsar.broker.resources.PulsarResources;
import org.apache.pulsar.broker.rest.Topics;
import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.HealthChecker;
import org.apache.pulsar.broker.service.PulsarMetadataEventSynchronizer;
import org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService;
import org.apache.pulsar.broker.service.Topic;
@@ -156,6 +155,7 @@ import org.apache.pulsar.common.configuration.VipStatus;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.naming.TopicVersion;
import org.apache.pulsar.common.policies.data.ClusterDataImpl;
import org.apache.pulsar.common.policies.data.OffloadPoliciesImpl;
import org.apache.pulsar.common.protocol.schema.SchemaStorage;
@@ -317,6 +317,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
private volatile CompletableFuture<Void> closeFuture;
// key is listener name, value is pulsar address and pulsar ssl address
private Map<String, AdvertisedListener> advertisedListeners;
+ private volatile HealthChecker healthChecker;
public PulsarService(ServiceConfiguration config) {
this(config, Optional.empty(), (exitCode) -> LOG.info("Process termination requested with code {}. "
@@ -447,41 +448,6 @@ public class PulsarService implements AutoCloseable, ShutdownService {
}
}
- private boolean isManagedLedgerNotFoundException(Throwable e) {
- Throwable realCause = e.getCause();
- return realCause instanceof ManagedLedgerException.MetadataNotFoundException
- || realCause instanceof MetadataStoreException.NotFoundException;
- }
-
- private void deleteHeartbeatResource() {
- if (this.brokerService != null) {
- LOG.info("forcefully delete heartbeat topic when close broker");
-
- String heartbeatTopicNameV1 = getHeartbeatTopicName(getBrokerId(), getConfiguration(), false);
- String heartbeatTopicNameV2 = getHeartbeatTopicName(getBrokerId(), getConfiguration(), true);
-
- try {
- this.brokerService.deleteTopic(heartbeatTopicNameV1, true).get();
- } catch (Exception e) {
- if (!isManagedLedgerNotFoundException(e)) {
- LOG.error("Closed with errors in delete heartbeat topic [{}]",
- heartbeatTopicNameV1, e);
- }
- }
-
- try {
- this.brokerService.deleteTopic(heartbeatTopicNameV2, true).get();
- } catch (Exception e) {
- if (!isManagedLedgerNotFoundException(e)) {
- LOG.error("Closed with errors in delete heartbeat topic [{}]",
- heartbeatTopicNameV2, e);
- }
- }
-
- LOG.info("finish forcefully delete heartbeat topic when close broker");
- }
- }
-
@Override
public void close() throws PulsarServerException {
try {
@@ -531,9 +497,9 @@ public class PulsarService implements AutoCloseable, ShutdownService {
// It only tells the Pulsar clients that this service is not ready to serve for the lookup requests
state = State.Closing;
- if (brokerId != null) {
- // forcefully delete heartbeat topic when close broker
- deleteHeartbeatResource();
+ if (healthChecker != null) {
+ healthChecker.close();
+ healthChecker = null;
}
// close the service in reverse order v.s. in which they are started
@@ -1699,78 +1665,36 @@ public class PulsarService implements AutoCloseable, ShutdownService {
return this.offloaderReadExecutor;
}
- public PulsarClientImpl createClientImpl(ClientConfigurationData clientConf)
+ public PulsarClientImpl createClientImpl(ClientConfigurationData conf) throws PulsarClientException {
+ return createClientImpl(conf, null);
+ }
+
+ public PulsarClientImpl createClientImpl(Consumer<PulsarClientImpl.PulsarClientImplBuilder> customizer)
+ throws PulsarClientException {
+ return createClientImpl(null, customizer);
+ }
+
+ public PulsarClientImpl createClientImpl(ClientConfigurationData conf,
+ Consumer<PulsarClientImpl.PulsarClientImplBuilder> customizer)
throws PulsarClientException {
- return PulsarClientImpl.builder()
- .conf(clientConf)
+ PulsarClientImpl.PulsarClientImplBuilder pulsarClientImplBuilder = PulsarClientImpl.builder()
+ .conf(conf != null ? conf : createClientConfigurationData())
.eventLoopGroup(ioEventLoopGroup)
.timer(brokerClientSharedTimer)
.internalExecutorProvider(brokerClientSharedInternalExecutorProvider)
.externalExecutorProvider(brokerClientSharedExternalExecutorProvider)
.scheduledExecutorProvider(brokerClientSharedScheduledExecutorProvider)
- .lookupExecutorProvider(brokerClientSharedLookupExecutorProvider)
- .build();
+ .lookupExecutorProvider(brokerClientSharedLookupExecutorProvider);
+ if (customizer != null) {
+ customizer.accept(pulsarClientImplBuilder);
+ }
+ return pulsarClientImplBuilder.build();
}
public synchronized PulsarClient getClient() throws PulsarServerException {
if (this.client == null) {
try {
- ClientConfigurationData initialConf = new ClientConfigurationData();
-
- // Disable memory limit for broker client and disable stats
- initialConf.setMemoryLimitBytes(0);
- initialConf.setStatsIntervalSeconds(0);
-
- // Apply all arbitrary configuration. This must be called before setting any fields annotated as
- // @Secret on the ClientConfigurationData object because of the way they are serialized.
- // See https://github.com/apache/pulsar/issues/8509 for more information.
- Map<String, Object> overrides = PropertiesUtils
- .filterAndMapProperties(this.getConfiguration().getProperties(), "brokerClient_");
- ClientConfigurationData conf =
- ConfigurationDataUtils.loadData(overrides, initialConf, ClientConfigurationData.class);
-
- // Disabled auto release useless connections
- // The automatic release connection feature is not yet perfect for transaction scenarios, so turn it
- // off first.
- conf.setConnectionMaxIdleSeconds(-1);
-
- boolean tlsEnabled = this.getConfiguration().isBrokerClientTlsEnabled();
- conf.setServiceUrl(tlsEnabled ? this.brokerServiceUrlTls : this.brokerServiceUrl);
-
- if (tlsEnabled) {
- conf.setTlsCiphers(this.getConfiguration().getBrokerClientTlsCiphers());
- conf.setTlsProtocols(this.getConfiguration().getBrokerClientTlsProtocols());
- conf.setTlsAllowInsecureConnection(this.getConfiguration().isTlsAllowInsecureConnection());
- conf.setTlsHostnameVerificationEnable(this.getConfiguration().isTlsHostnameVerificationEnabled());
- conf.setSslFactoryPlugin(this.getConfiguration().getBrokerClientSslFactoryPlugin());
- conf.setSslFactoryPluginParams(this.getConfiguration().getBrokerClientSslFactoryPluginParams());
- if (this.getConfiguration().isBrokerClientTlsEnabledWithKeyStore()) {
- conf.setUseKeyStoreTls(true);
- conf.setTlsTrustStoreType(this.getConfiguration().getBrokerClientTlsTrustStoreType());
- conf.setTlsTrustStorePath(this.getConfiguration().getBrokerClientTlsTrustStore());
- conf.setTlsTrustStorePassword(this.getConfiguration().getBrokerClientTlsTrustStorePassword());
- conf.setTlsKeyStoreType(this.getConfiguration().getBrokerClientTlsKeyStoreType());
- conf.setTlsKeyStorePath(this.getConfiguration().getBrokerClientTlsKeyStore());
- conf.setTlsKeyStorePassword(this.getConfiguration().getBrokerClientTlsKeyStorePassword());
- } else {
- conf.setTlsTrustCertsFilePath(
- isNotBlank(this.getConfiguration().getBrokerClientTrustCertsFilePath())
- ? this.getConfiguration().getBrokerClientTrustCertsFilePath()
- : this.getConfiguration().getTlsTrustCertsFilePath());
- conf.setTlsKeyFilePath(this.getConfiguration().getBrokerClientKeyFilePath());
- conf.setTlsCertificateFilePath(this.getConfiguration().getBrokerClientCertificateFilePath());
- }
- }
-
- if (isNotBlank(this.getConfiguration().getBrokerClientAuthenticationPlugin())) {
- conf.setAuthPluginClassName(this.getConfiguration().getBrokerClientAuthenticationPlugin());
- conf.setAuthParams(this.getConfiguration().getBrokerClientAuthenticationParameters());
- conf.setAuthParamMap(null);
- conf.setAuthentication(AuthenticationFactory.create(
- this.getConfiguration().getBrokerClientAuthenticationPlugin(),
- this.getConfiguration().getBrokerClientAuthenticationParameters()));
- }
- this.client = createClientImpl(conf);
+ this.client = createClientImpl(null, null);
} catch (Exception e) {
throw new PulsarServerException(e);
}
@@ -1778,63 +1702,126 @@ public class PulsarService implements AutoCloseable, ShutdownService {
return this.client;
}
+ protected ClientConfigurationData createClientConfigurationData()
+ throws PulsarClientException.UnsupportedAuthenticationException {
+ ClientConfigurationData initialConf = new ClientConfigurationData();
+
+ // Disable memory limit for broker client and disable stats
+ initialConf.setMemoryLimitBytes(0);
+ initialConf.setStatsIntervalSeconds(0);
+
+ // Apply all arbitrary configuration. This must be called before setting any fields annotated as
+ // @Secret on the ClientConfigurationData object because of the way they are serialized.
+ // See https://github.com/apache/pulsar/issues/8509 for more information.
+ Map<String, Object> overrides = PropertiesUtils
+ .filterAndMapProperties(this.getConfiguration().getProperties(), "brokerClient_");
+ ClientConfigurationData conf =
+ ConfigurationDataUtils.loadData(overrides, initialConf, ClientConfigurationData.class);
+
+ // Disabled auto release useless connections
+ // The automatic release connection feature is not yet perfect for transaction scenarios, so turn it
+ // off first.
+ conf.setConnectionMaxIdleSeconds(-1);
+
+ boolean tlsEnabled = this.getConfiguration().isBrokerClientTlsEnabled();
+ conf.setServiceUrl(tlsEnabled ? this.brokerServiceUrlTls : this.brokerServiceUrl);
+
+ if (tlsEnabled) {
+ conf.setTlsCiphers(this.getConfiguration().getBrokerClientTlsCiphers());
+ conf.setTlsProtocols(this.getConfiguration().getBrokerClientTlsProtocols());
+ conf.setTlsAllowInsecureConnection(this.getConfiguration().isTlsAllowInsecureConnection());
+ conf.setTlsHostnameVerificationEnable(this.getConfiguration().isTlsHostnameVerificationEnabled());
+ conf.setSslFactoryPlugin(this.getConfiguration().getBrokerClientSslFactoryPlugin());
+ conf.setSslFactoryPluginParams(this.getConfiguration().getBrokerClientSslFactoryPluginParams());
+ if (this.getConfiguration().isBrokerClientTlsEnabledWithKeyStore()) {
+ conf.setUseKeyStoreTls(true);
+ conf.setTlsTrustStoreType(this.getConfiguration().getBrokerClientTlsTrustStoreType());
+ conf.setTlsTrustStorePath(this.getConfiguration().getBrokerClientTlsTrustStore());
+ conf.setTlsTrustStorePassword(this.getConfiguration().getBrokerClientTlsTrustStorePassword());
+ conf.setTlsKeyStoreType(this.getConfiguration().getBrokerClientTlsKeyStoreType());
+ conf.setTlsKeyStorePath(this.getConfiguration().getBrokerClientTlsKeyStore());
+ conf.setTlsKeyStorePassword(this.getConfiguration().getBrokerClientTlsKeyStorePassword());
+ } else {
+ conf.setTlsTrustCertsFilePath(
+ isNotBlank(this.getConfiguration().getBrokerClientTrustCertsFilePath())
+ ? this.getConfiguration().getBrokerClientTrustCertsFilePath()
+ : this.getConfiguration().getTlsTrustCertsFilePath());
+ conf.setTlsKeyFilePath(this.getConfiguration().getBrokerClientKeyFilePath());
+ conf.setTlsCertificateFilePath(this.getConfiguration().getBrokerClientCertificateFilePath());
+ }
+ }
+
+ if (isNotBlank(this.getConfiguration().getBrokerClientAuthenticationPlugin())) {
+ conf.setAuthPluginClassName(this.getConfiguration().getBrokerClientAuthenticationPlugin());
+ conf.setAuthParams(this.getConfiguration().getBrokerClientAuthenticationParameters());
+ conf.setAuthParamMap(null);
+ conf.setAuthentication(AuthenticationFactory.create(
+ this.getConfiguration().getBrokerClientAuthenticationPlugin(),
+ this.getConfiguration().getBrokerClientAuthenticationParameters()));
+ }
+ return conf;
+ }
+
public synchronized PulsarAdmin getAdminClient() throws PulsarServerException {
if (this.adminClient == null) {
try {
- ServiceConfiguration conf = this.getConfiguration();
- final String adminApiUrl = conf.isBrokerClientTlsEnabled() ? webServiceAddressTls : webServiceAddress;
- if (adminApiUrl == null) {
- throw new IllegalArgumentException("Web service address was not set properly "
- + ", isBrokerClientTlsEnabled: " + conf.isBrokerClientTlsEnabled()
- + ", webServiceAddressTls: " + webServiceAddressTls
- + ", webServiceAddress: " + webServiceAddress);
- }
- PulsarAdminBuilder builder = PulsarAdmin.builder().serviceHttpUrl(adminApiUrl);
-
- // Apply all arbitrary configuration. This must be called before setting any fields annotated as
- // @Secret on the ClientConfigurationData object because of the way they are serialized.
- // See https://github.com/apache/pulsar/issues/8509 for more information.
- builder.loadConf(PropertiesUtils.filterAndMapProperties(conf.getProperties(), "brokerClient_"));
-
- builder.authentication(
- conf.getBrokerClientAuthenticationPlugin(),
- conf.getBrokerClientAuthenticationParameters());
-
- if (conf.isBrokerClientTlsEnabled()) {
- builder.tlsCiphers(conf.getBrokerClientTlsCiphers())
- .tlsProtocols(conf.getBrokerClientTlsProtocols())
- .sslFactoryPlugin(conf.getBrokerClientSslFactoryPlugin())
- .sslFactoryPluginParams(conf.getBrokerClientSslFactoryPluginParams());
- if (conf.isBrokerClientTlsEnabledWithKeyStore()) {
- builder.useKeyStoreTls(true).tlsTrustStoreType(conf.getBrokerClientTlsTrustStoreType())
- .tlsTrustStorePath(conf.getBrokerClientTlsTrustStore())
- .tlsTrustStorePassword(conf.getBrokerClientTlsTrustStorePassword())
- .tlsKeyStoreType(conf.getBrokerClientTlsKeyStoreType())
- .tlsKeyStorePath(conf.getBrokerClientTlsKeyStore())
- .tlsKeyStorePassword(conf.getBrokerClientTlsKeyStorePassword());
- } else {
- builder.tlsTrustCertsFilePath(conf.getBrokerClientTrustCertsFilePath())
- .tlsKeyFilePath(conf.getBrokerClientKeyFilePath())
- .tlsCertificateFilePath(conf.getBrokerClientCertificateFilePath());
- }
- builder.allowTlsInsecureConnection(conf.isTlsAllowInsecureConnection())
- .enableTlsHostnameVerification(conf.isTlsHostnameVerificationEnabled());
- }
-
- // most of the admin request requires to make zk-call so, keep the max read-timeout based on
- // zk-operation timeout
- builder.readTimeout(conf.getMetadataStoreOperationTimeoutSeconds(), TimeUnit.SECONDS);
-
- this.adminClient = builder.build();
- LOG.info("created admin with url {} ", adminApiUrl);
+ this.adminClient = getCreateAdminClientBuilder().build();
+ LOG.info("created admin with url {} ", adminClient.getServiceUrl());
} catch (Exception e) {
throw new PulsarServerException(e);
}
}
-
return this.adminClient;
}
+ protected PulsarAdminBuilder getCreateAdminClientBuilder()
+ throws PulsarClientException.UnsupportedAuthenticationException {
+ ServiceConfiguration conf = this.getConfiguration();
+ final String adminApiUrl = conf.isBrokerClientTlsEnabled() ? webServiceAddressTls : webServiceAddress;
+ if (adminApiUrl == null) {
+ throw new IllegalArgumentException("Web service address was not set properly "
+ + ", isBrokerClientTlsEnabled: " + conf.isBrokerClientTlsEnabled()
+ + ", webServiceAddressTls: " + webServiceAddressTls
+ + ", webServiceAddress: " + webServiceAddress);
+ }
+ PulsarAdminBuilder builder = PulsarAdmin.builder().serviceHttpUrl(adminApiUrl);
+
+ // Apply all arbitrary configuration. This must be called before setting any fields annotated as
+ // @Secret on the ClientConfigurationData object because of the way they are serialized.
+ // See https://github.com/apache/pulsar/issues/8509 for more information.
+ builder.loadConf(PropertiesUtils.filterAndMapProperties(conf.getProperties(), "brokerClient_"));
+
+ builder.authentication(
+ conf.getBrokerClientAuthenticationPlugin(),
+ conf.getBrokerClientAuthenticationParameters());
+
+ if (conf.isBrokerClientTlsEnabled()) {
+ builder.tlsCiphers(conf.getBrokerClientTlsCiphers())
+ .tlsProtocols(conf.getBrokerClientTlsProtocols())
+ .sslFactoryPlugin(conf.getBrokerClientSslFactoryPlugin())
+ .sslFactoryPluginParams(conf.getBrokerClientSslFactoryPluginParams());
+ if (conf.isBrokerClientTlsEnabledWithKeyStore()) {
+ builder.useKeyStoreTls(true).tlsTrustStoreType(conf.getBrokerClientTlsTrustStoreType())
+ .tlsTrustStorePath(conf.getBrokerClientTlsTrustStore())
+ .tlsTrustStorePassword(conf.getBrokerClientTlsTrustStorePassword())
+ .tlsKeyStoreType(conf.getBrokerClientTlsKeyStoreType())
+ .tlsKeyStorePath(conf.getBrokerClientTlsKeyStore())
+ .tlsKeyStorePassword(conf.getBrokerClientTlsKeyStorePassword());
+ } else {
+ builder.tlsTrustCertsFilePath(conf.getBrokerClientTrustCertsFilePath())
+ .tlsKeyFilePath(conf.getBrokerClientKeyFilePath())
+ .tlsCertificateFilePath(conf.getBrokerClientCertificateFilePath());
+ }
+ builder.allowTlsInsecureConnection(conf.isTlsAllowInsecureConnection())
+ .enableTlsHostnameVerification(conf.isTlsHostnameVerificationEnabled());
+ }
+
+ // most of the admin request requires to make zk-call so, keep the max read-timeout based on
+ // zk-operation timeout
+ builder.readTimeout(conf.getMetadataStoreOperationTimeoutSeconds(), TimeUnit.SECONDS);
+ return builder;
+ }
+
/**
* Gets the broker service URL (non-TLS) associated with the internal listener.
*/
@@ -2181,4 +2168,40 @@ public class PulsarService implements AutoCloseable, ShutdownService {
return (TopicPoliciesService) Reflections.createInstance(className,
Thread.currentThread().getContextClassLoader());
}
+
+ /**
+ * Run health check for the broker.
+ *
+ * @return CompletableFuture
+ */
+ public CompletableFuture<Void> runHealthCheck(TopicVersion topicVersion, String clientId) {
+ if (!isRunning()) {
+ return CompletableFuture.failedFuture(new PulsarServerException("Broker is not running"));
+ }
+ HealthChecker localHealthChecker = getHealthChecker();
+ if (localHealthChecker == null) {
+ return CompletableFuture.failedFuture(new PulsarServerException("Broker is not running"));
+ }
+ return localHealthChecker.checkHealth(topicVersion, clientId);
+ }
+
+ @VisibleForTesting
+ public HealthChecker getHealthChecker() {
+ if (healthChecker == null) {
+ synchronized (this) {
+ if (healthChecker == null) {
+ if (!isRunning()) {
+ return null;
+ }
+ try {
+ healthChecker = new HealthChecker(this);
+ } catch (PulsarServerException e) {
+ LOG.error("Failed to create health checker", e);
+ throw new RuntimeException(e);
+ }
+ }
+ }
+ }
+ return healthChecker;
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
index a24a78d8e31..ea25367ca3a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
@@ -26,16 +26,11 @@ import io.swagger.annotations.ApiResponses;
import java.lang.management.ManagementFactory;
import java.lang.management.ThreadInfo;
import java.lang.management.ThreadMXBean;
-import java.time.Duration;
-import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
-import java.util.UUID;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import javax.ws.rs.DELETE;
import javax.ws.rs.DefaultValue;
@@ -50,23 +45,12 @@ import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;
import org.apache.commons.lang.StringUtils;
import org.apache.pulsar.PulsarVersion;
-import org.apache.pulsar.broker.PulsarServerException;
-import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.PulsarService.State;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.admin.AdminResource;
import org.apache.pulsar.broker.loadbalance.LeaderBroker;
-import org.apache.pulsar.broker.namespace.NamespaceService;
-import org.apache.pulsar.broker.service.Subscription;
-import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.web.RestException;
-import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.api.Producer;
-import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.Reader;
-import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.conf.InternalConfigurationData;
-import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicVersion;
import org.apache.pulsar.common.policies.data.BrokerInfo;
import org.apache.pulsar.common.policies.data.BrokerOperation;
@@ -81,16 +65,9 @@ import org.slf4j.LoggerFactory;
*/
public class BrokersBase extends AdminResource {
private static final Logger LOG = LoggerFactory.getLogger(BrokersBase.class);
- public static final String HEALTH_CHECK_TOPIC_SUFFIX = "healthcheck";
// log a full thread dump when a deadlock is detected in healthcheck once every 10 minutes
// to prevent excessive logging
private static final long LOG_THREADDUMP_INTERVAL_WHEN_DEADLOCK_DETECTED = 600000L;
- // there is a timeout of 60 seconds default in the client(readTimeoutMs), so we need to set the timeout
- // a bit shorter than 60 seconds to avoid the client timeout exception thrown before the server timeout exception.
- // or we can't propagate the server timeout exception to the client.
- private static final Duration HEALTH_CHECK_READ_TIMEOUT = Duration.ofSeconds(58);
- private static final TimeoutException HEALTH_CHECK_TIMEOUT_EXCEPTION =
- FutureUtil.createTimeoutException("Timeout", BrokersBase.class, "healthCheckRecursiveReadNext(...)");
private static volatile long threadDumpLoggedTimestamp;
@GET
@@ -385,16 +362,21 @@ public class BrokersBase extends AdminResource {
@ApiResponse(code = 307, message = "Current broker is not the target broker"),
@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Cluster doesn't exist"),
- @ApiResponse(code = 500, message = "Internal server error")})
+ @ApiResponse(code = 500, message = "Internal server error"),
+ @ApiResponse(code = 503, message = "Service unavailable")})
public void healthCheck(@Suspended AsyncResponse asyncResponse,
@ApiParam(value = "Topic Version")
@QueryParam("topicVersion") TopicVersion topicVersion,
@QueryParam("brokerId") String brokerId) {
+ if (pulsar().getState() == State.Closed || pulsar().getState() == State.Closing) {
+ asyncResponse.resume(Response.status(Status.SERVICE_UNAVAILABLE).build());
+ return;
+ }
validateBothSuperuserAndBrokerOperation(pulsar().getConfig().getClusterName(), StringUtils.isBlank(brokerId)
? pulsar().getBrokerId() : brokerId, BrokerOperation.HEALTH_CHECK)
- .thenAccept(__ -> checkDeadlockedThreads())
.thenCompose(__ -> maybeRedirectToBroker(
StringUtils.isBlank(brokerId) ? pulsar().getBrokerId() : brokerId))
+ .thenAccept(__ -> checkDeadlockedThreads())
.thenCompose(__ -> internalRunHealthCheck(topicVersion))
.thenAccept(__ -> {
LOG.info("[{}] Successfully run health check.", clientAppId());
@@ -432,143 +414,8 @@ public class BrokersBase extends AdminResource {
}
}
- public static String getHeartbeatTopicName(String brokerId, ServiceConfiguration configuration, boolean isV2) {
- NamespaceName namespaceName = isV2
- ? NamespaceService.getHeartbeatNamespaceV2(brokerId, configuration)
- : NamespaceService.getHeartbeatNamespace(brokerId, configuration);
- return String.format("persistent://%s/%s", namespaceName, HEALTH_CHECK_TOPIC_SUFFIX);
- }
-
private CompletableFuture<Void> internalRunHealthCheck(TopicVersion topicVersion) {
- return internalRunHealthCheck(topicVersion, pulsar(), clientAppId());
- }
-
-
- public static CompletableFuture<Void> internalRunHealthCheck(TopicVersion topicVersion, PulsarService pulsar,
- String clientAppId) {
- NamespaceName namespaceName = (topicVersion == TopicVersion.V2)
- ? NamespaceService.getHeartbeatNamespaceV2(pulsar.getAdvertisedAddress(), pulsar.getConfiguration())
- : NamespaceService.getHeartbeatNamespace(pulsar.getAdvertisedAddress(), pulsar.getConfiguration());
- String brokerId = pulsar.getBrokerId();
- final String topicName =
- getHeartbeatTopicName(brokerId, pulsar.getConfiguration(), (topicVersion == TopicVersion.V2));
- LOG.info("[{}] Running healthCheck with topic={}", clientAppId, topicName);
- final String messageStr = UUID.randomUUID().toString();
- final String subscriptionName = "healthCheck-" + messageStr;
- // create non-partitioned topic manually and close the previous reader if present.
- return pulsar.getBrokerService().getTopic(topicName, true)
- .thenCompose(topicOptional -> {
- if (!topicOptional.isPresent()) {
- LOG.error("[{}] Fail to run health check while get topic {}. because get null value.",
- clientAppId, topicName);
- throw new RestException(Status.NOT_FOUND,
- String.format("Topic [%s] not found after create.", topicName));
- }
- PulsarClient client;
- try {
- client = pulsar.getClient();
- } catch (PulsarServerException e) {
- LOG.error("[{}] Fail to run health check while get client.", clientAppId);
- throw new RestException(e);
- }
- CompletableFuture<Void> resultFuture = new CompletableFuture<>();
- client.newProducer(Schema.STRING).topic(topicName).createAsync()
- .thenCompose(producer -> client.newReader(Schema.STRING).topic(topicName)
- .subscriptionName(subscriptionName)
- .startMessageId(MessageId.latest)
- .createAsync().exceptionally(createException -> {
- producer.closeAsync().exceptionally(ex -> {
- LOG.error("[{}] Close producer fail while heath check.", clientAppId);
- return null;
- });
- throw FutureUtil.wrapToCompletionException(createException);
- }).thenCompose(reader -> producer.sendAsync(messageStr)
- .thenCompose(__ -> FutureUtil.addTimeoutHandling(
- healthCheckRecursiveReadNext(reader, messageStr),
- HEALTH_CHECK_READ_TIMEOUT, pulsar.getBrokerService().executor(),
- () -> HEALTH_CHECK_TIMEOUT_EXCEPTION))
- .whenComplete((__, ex) -> {
- closeAndReCheck(producer, reader, topicOptional.get(), subscriptionName,
- clientAppId)
- .whenComplete((unused, innerEx) -> {
- if (ex != null) {
- resultFuture.completeExceptionally(ex);
- } else {
- resultFuture.complete(null);
- }
- });
- }
- ))
- ).exceptionally(ex -> {
- resultFuture.completeExceptionally(ex);
- return null;
- });
- return resultFuture;
- });
- }
-
- private CompletableFuture<Void> closeAndReCheck(Producer<String> producer, Reader<String> reader,
- Topic topic, String subscriptionName) {
- return closeAndReCheck(producer, reader, topic, subscriptionName, clientAppId());
- }
-
- /**
- * Close producer and reader and then to re-check if this operation is success.
- *
- * Re-check
- * - Producer: If close fails we will print error log to notify user.
- * - Consumer: If close fails we will force delete subscription.
- *
- * @param producer Producer
- * @param reader Reader
- * @param topic Topic
- * @param subscriptionName Subscription name
- */
- private static CompletableFuture<Void> closeAndReCheck(Producer<String> producer, Reader<String> reader,
- Topic topic, String subscriptionName, String clientAppId) {
- // no matter exception or success, we still need to
- // close producer/reader
- CompletableFuture<Void> producerFuture = producer.closeAsync();
- CompletableFuture<Void> readerFuture = reader.closeAsync();
- List<CompletableFuture<Void>> futures = new ArrayList<>(2);
- futures.add(producerFuture);
- futures.add(readerFuture);
- return FutureUtil.waitForAll(Collections.unmodifiableList(futures))
- .exceptionally(closeException -> {
- if (readerFuture.isCompletedExceptionally()) {
- LOG.error("[{}] Close reader fail while heath check.", clientAppId);
- Subscription subscription =
- topic.getSubscription(subscriptionName);
- // re-check subscription after reader close
- if (subscription != null) {
- LOG.warn("[{}] Force delete subscription {} "
- + "when it still exists after the"
- + " reader is closed.",
- clientAppId, subscription);
- subscription.deleteForcefully()
- .exceptionally(ex -> {
- LOG.error("[{}] Force delete subscription fail"
- + " while health check",
- clientAppId, ex);
- return null;
- });
- }
- } else {
- // producer future fail.
- LOG.error("[{}] Close producer fail while heath check.", clientAppId);
- }
- return null;
- });
- }
-
- private static CompletableFuture<Void> healthCheckRecursiveReadNext(Reader<String> reader, String content) {
- return reader.readNextAsync()
- .thenCompose(msg -> {
- if (!Objects.equals(content, msg.getValue())) {
- return healthCheckRecursiveReadNext(reader, content);
- }
- return CompletableFuture.completedFuture(null);
- });
+ return pulsar().runHealthCheck(topicVersion, clientAppId());
}
private CompletableFuture<Void> internalDeleteDynamicConfigurationOnMetadataAsync(String configName) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index dd3486496e9..9f6f3dacbbe 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -24,7 +24,6 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.bookkeeper.mledger.ManagedLedgerConfig.PROPERTY_SOURCE_TOPIC_KEY;
import static org.apache.commons.collections4.CollectionUtils.isEmpty;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
-import static org.apache.pulsar.broker.admin.impl.BrokersBase.internalRunHealthCheck;
import static org.apache.pulsar.client.util.RetryMessageUtil.DLQ_GROUP_TOPIC_SUFFIX;
import static org.apache.pulsar.client.util.RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX;
import static org.apache.pulsar.common.naming.SystemTopicNames.isTransactionInternalName;
@@ -679,7 +678,10 @@ public class BrokerService implements Closeable {
}
public CompletableFuture<Void> checkHealth() {
- return internalRunHealthCheck(TopicVersion.V2, pulsar(), null).thenAccept(__ -> {
+ if (!pulsar().isRunning()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ return pulsar().runHealthCheck(TopicVersion.V2, null).thenAccept(__ -> {
this.pulsarStats.getBrokerOperabilityMetrics().recordHealthCheckStatusSuccess();
}).exceptionally(ex -> {
this.pulsarStats.getBrokerOperabilityMetrics().recordHealthCheckStatusFail();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/HealthChecker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/HealthChecker.java
new file mode 100644
index 00000000000..e80e0b09475
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/HealthChecker.java
@@ -0,0 +1,330 @@
+/*
+ * 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.pulsar.broker.service;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeoutException;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.pulsar.broker.PulsarServerException;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.namespace.NamespaceService;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.util.ScheduledExecutorProvider;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.naming.TopicVersion;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.metadata.api.MetadataStoreException;
+
+/**
+ * The HealthChecker class provides functionality to monitor and verify the health of a Pulsar broker.
+ * It performs health checks by creating test topics, producing and consuming messages to verify broker functionality.
+ * This class implements AutoCloseable to ensure proper cleanup of resources when the broker is shut down.
+ */
+@Slf4j
+public class HealthChecker implements AutoCloseable{
+ /**
+ * Suffix used for health check topic names.
+ */
+ public static final String HEALTH_CHECK_TOPIC_SUFFIX = "healthcheck";
+ /**
+ * Timeout duration for health check operations.
+ * Set to 58 seconds to be shorter than the client's default 60-second timeout,
+ * allowing server timeout exceptions to propagate properly to the client.
+ */
+ private static final Duration DEFAULT_HEALTH_CHECK_READ_TIMEOUT = Duration.ofSeconds(58);
+ /**
+ * Pre-created timeout exception for health check operations.
+ */
+ private static final TimeoutException HEALTH_CHECK_TIMEOUT_EXCEPTION =
+ FutureUtil.createTimeoutException("Timeout", HealthChecker.class, "healthCheckRecursiveReadNext(...)");
+ /**
+ * Reference to the main Pulsar service.
+ */
+ private final PulsarService pulsar;
+ /**
+ * Topic name for v1 heartbeat checks.
+ */
+ private final String heartbeatTopicV1;
+ /**
+ * Topic name for v2 heartbeat checks.
+ */
+ private final String heartbeatTopicV2;
+ /**
+ * Pulsar client instance for health check operations.
+ * A separate client is needed so that it can be shutdown before the webservice is closed.
+ * Pending requests for healthchecks to the /health endpoint can be cancelled this way.
+ */
+ private final PulsarClient client;
+ /**
+ * Executor for lookup operations.
+ * This is also needed so that pending healthchecks can be properly cancelled at shutdown.
+ */
+ private final ScheduledExecutorProvider lookupExecutor;
+ /**
+ * Executor for scheduled tasks.
+ * This is also needed so that pending healthchecks can be properly cancelled at shutdown.
+ */
+ private final ScheduledExecutorProvider scheduledExecutorProvider;
+ /**
+ * Set of pending health check operations.
+ */
+ private final Set<CompletableFuture<Void>> pendingFutures = new HashSet<>();
+
+ private final Duration timeout = DEFAULT_HEALTH_CHECK_READ_TIMEOUT;
+
+ public HealthChecker(PulsarService pulsar) throws PulsarServerException {
+ this.pulsar = pulsar;
+ this.heartbeatTopicV1 = getHeartbeatTopicName(pulsar.getBrokerId(), pulsar.getConfiguration(), false);
+ this.heartbeatTopicV2 = getHeartbeatTopicName(pulsar.getBrokerId(), pulsar.getConfiguration(), true);
+ this.lookupExecutor =
+ new ScheduledExecutorProvider(1, "health-checker-client-lookup-executor");
+ this.scheduledExecutorProvider =
+ new ScheduledExecutorProvider(1, "health-checker-client-scheduled-executor");
+ try {
+ this.client = pulsar.createClientImpl(builder -> {
+ builder.lookupExecutorProvider(lookupExecutor);
+ builder.scheduledExecutorProvider(scheduledExecutorProvider);
+ });
+ } catch (PulsarClientException e) {
+ throw new PulsarServerException("Error creating client for HealthChecker", e);
+ }
+ }
+
+ private static String getHeartbeatTopicName(String brokerId, ServiceConfiguration configuration, boolean isV2) {
+ NamespaceName namespaceName = isV2
+ ? NamespaceService.getHeartbeatNamespaceV2(brokerId, configuration)
+ : NamespaceService.getHeartbeatNamespace(brokerId, configuration);
+ return String.format("persistent://%s/%s", namespaceName, HEALTH_CHECK_TOPIC_SUFFIX);
+ }
+
+ /**
+ * Performs a health check on the broker by verifying message production and consumption.
+ * The health check process includes:
+ * 1. Producing a test message
+ * 2. Reading the message back to verify end-to-end functionality
+ *
+ * @param topicVersion The version of the topic to use (V1 or V2)
+ * @param clientAppId The identifier of the client application requesting the health check
+ * @return A CompletableFuture that completes when the health check is successful, or completes exceptionally if the
+ * check fails
+ */
+ public CompletableFuture<Void> checkHealth(TopicVersion topicVersion, String clientAppId) {
+ final String topicName = topicVersion == TopicVersion.V2 ? heartbeatTopicV2 : heartbeatTopicV1;
+ log.info("[{}] Running healthCheck with topic={}", clientAppId, topicName);
+ final String messageStr = UUID.randomUUID().toString();
+ final String subscriptionName = "healthCheck-" + messageStr;
+ // create non-partitioned topic manually and close the previous reader if present.
+ CompletableFuture<Void> resultFuture = new CompletableFuture<>();
+ addToPending(resultFuture);
+ resultFuture.whenComplete((result, ex) -> {
+ removeFromPending(resultFuture);
+ });
+ try {
+ pulsar.getBrokerService().getTopic(topicName, true)
+ .thenCompose(topicOptional -> {
+ if (!topicOptional.isPresent()) {
+ log.error("[{}] Fail to run health check while get topic {}. because get null value.",
+ clientAppId, topicName);
+ return CompletableFuture.failedFuture(new BrokerServiceException.TopicNotFoundException(
+ String.format("Topic [%s] not found after create.", topicName)));
+ }
+ return doHealthCheck(clientAppId, topicName, subscriptionName, messageStr, resultFuture);
+ }).whenComplete((result, t) -> {
+ if (t != null) {
+ resultFuture.completeExceptionally(t);
+ } else {
+ if (!resultFuture.isDone()) {
+ resultFuture.complete(null);
+ }
+ }
+ });
+ } catch (Exception e) {
+ log.error("[{}] Fail to run health check while get topic {}. because get exception.",
+ clientAppId, topicName, e);
+ resultFuture.completeExceptionally(e);
+ }
+ return resultFuture;
+ }
+
+ private synchronized void addToPending(CompletableFuture<Void> resultFuture) {
+ pendingFutures.add(resultFuture);
+ }
+
+ private synchronized void removeFromPending(CompletableFuture<Void> resultFuture) {
+ pendingFutures.remove(resultFuture);
+ }
+
+ private CompletableFuture<Void> doHealthCheck(String clientAppId, String topicName, String subscriptionName,
+ String messageStr, CompletableFuture<Void> resultFuture) {
+ return client.newProducer(Schema.STRING).topic(topicName).createAsync()
+ .thenCompose(producer -> client.newReader(Schema.STRING).topic(topicName)
+ .subscriptionName(subscriptionName)
+ .startMessageId(MessageId.latest)
+ .createAsync().exceptionally(createException -> {
+ producer.closeAsync().exceptionally(ex -> {
+ log.error("[{}] Close producer fail while heath check.", clientAppId);
+ return null;
+ });
+ throw FutureUtil.wrapToCompletionException(createException);
+ }).thenCompose(reader -> producer.sendAsync(messageStr)
+ .thenCompose(__ -> FutureUtil.addTimeoutHandling(
+ healthCheckRecursiveReadNext(reader, messageStr),
+ timeout, pulsar.getBrokerService().executor(),
+ () -> HEALTH_CHECK_TIMEOUT_EXCEPTION))
+ .whenComplete((__, ex) -> {
+ closeAndReCheck(producer, reader, topicName,
+ subscriptionName,
+ clientAppId)
+ .whenComplete((unused, innerEx) -> {
+ if (ex != null) {
+ resultFuture.completeExceptionally(ex);
+ } else {
+ resultFuture.complete(null);
+ }
+ });
+ }
+ ))
+ ).exceptionally(ex -> {
+ resultFuture.completeExceptionally(ex);
+ return null;
+ });
+ }
+
+ /**
+ * Close producer and reader and then to re-check if this operation is success.
+ *
+ * Re-check
+ * - Producer: If close fails we will print error log to notify user.
+ * - Consumer: If close fails we will force delete subscription.
+ *
+ * @param producer Producer
+ * @param reader Reader
+ * @param subscriptionName Subscription name
+ */
+ private CompletableFuture<Void> closeAndReCheck(Producer<String> producer, Reader<String> reader,
+ String topicName, String subscriptionName, String clientAppId) {
+ // no matter exception or success, we still need to
+ // close producer/reader
+ CompletableFuture<Void> producerFuture = producer.closeAsync();
+ CompletableFuture<Void> readerFuture = reader.closeAsync();
+ List<CompletableFuture<Void>> futures = new ArrayList<>(2);
+ futures.add(producerFuture);
+ futures.add(readerFuture);
+ return FutureUtil.waitForAll(Collections.unmodifiableList(futures))
+ .exceptionally(closeException -> {
+ if (readerFuture.isCompletedExceptionally()) {
+ log.error("[{}] Close reader fail while health check.", clientAppId);
+ Optional<Topic> topic = pulsar.getBrokerService().getTopicReference(topicName);
+ if (topic.isPresent()) {
+ Subscription subscription =
+ topic.get().getSubscription(subscriptionName);
+ // re-check subscription after reader close
+ if (subscription != null) {
+ log.warn("[{}] Force delete subscription {} "
+ + "when it still exists after the"
+ + " reader is closed.",
+ clientAppId, subscription);
+ subscription.deleteForcefully()
+ .exceptionally(ex -> {
+ log.error("[{}] Force delete subscription fail"
+ + " while health check",
+ clientAppId, ex);
+ return null;
+ });
+ }
+ }
+ } else {
+ // producer future fail.
+ log.error("[{}] Close producer fail while heath check.", clientAppId);
+ }
+ return null;
+ });
+ }
+
+ private static CompletableFuture<Void> healthCheckRecursiveReadNext(Reader<String> reader, String content) {
+ return reader.readNextAsync()
+ .thenCompose(msg -> {
+ if (!Objects.equals(content, msg.getValue())) {
+ return healthCheckRecursiveReadNext(reader, content);
+ }
+ return CompletableFuture.completedFuture(null);
+ });
+ }
+
+ private void deleteHeartbeatTopics() {
+ log.info("forcefully deleting heartbeat topics");
+ deleteTopic(heartbeatTopicV1);
+ deleteTopic(heartbeatTopicV2);
+ log.info("finish forcefully deleting heartbeat topics");
+ }
+
+ private void deleteTopic(String heartbeatTopicV1) {
+ try {
+ pulsar.getBrokerService().deleteTopic(heartbeatTopicV1, true).get();
+ } catch (Exception e) {
+ Throwable realCause = e.getCause();
+ if (!(realCause instanceof ManagedLedgerException.MetadataNotFoundException
+ || realCause instanceof MetadataStoreException.NotFoundException)) {
+ log.error("Errors in deleting heartbeat topic [{}]", heartbeatTopicV1, e);
+ }
+ }
+ }
+
+ @Override
+ public synchronized void close() throws Exception {
+ try {
+ scheduledExecutorProvider.shutdownNow();
+ } catch (Exception e) {
+ log.warn("Failed to shutdown scheduled executor", e);
+ }
+ try {
+ lookupExecutor.shutdownNow();
+ } catch (Exception e) {
+ log.warn("Failed to shutdown lookup executor", e);
+ }
+ try {
+ client.close();
+ } catch (PulsarClientException e) {
+ log.warn("Failed to close pulsar client", e);
+ }
+ for (CompletableFuture<Void> pendingFuture : new ArrayList<>(pendingFutures)) {
+ if (!pendingFuture.isDone()) {
+ pendingFuture.completeExceptionally(
+ new PulsarClientException.AlreadyClosedException("HealthChecker is closed"));
+ }
+ }
+ deleteHeartbeatTopics();
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiHealthCheckTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiHealthCheckTest.java
index 618e023ccbf..b39f1f955a3 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiHealthCheckTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiHealthCheckTest.java
@@ -18,10 +18,11 @@
*/
package org.apache.pulsar.broker.admin;
-import static org.apache.pulsar.broker.admin.impl.BrokersBase.HEALTH_CHECK_TOPIC_SUFFIX;
+import static org.apache.pulsar.broker.service.HealthChecker.HEALTH_CHECK_TOPIC_SUFFIX;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
import java.lang.management.ManagementFactory;
import java.lang.management.ThreadMXBean;
import java.lang.reflect.Field;
@@ -33,9 +34,9 @@ import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
-import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.broker.namespace.NamespaceService;
+import org.apache.pulsar.broker.service.HealthChecker;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
@@ -231,17 +232,23 @@ public class AdminApiHealthCheckTest extends MockedPulsarServiceBaseTest {
public void testHealthCheckTimeOut() throws Exception {
final String testHealthCheckTopic = String.format("persistent://pulsar/localhost:%s/healthcheck",
pulsar.getConfig().getWebServicePort().get());
- PulsarClient client = pulsar.getClient();
+ HealthChecker healthChecker = pulsar.getHealthChecker();
+ Field clientField = HealthChecker.class.getDeclaredField("client");
+ clientField.setAccessible(true);
+ PulsarClient client = (PulsarClient) clientField.get(healthChecker);
PulsarClient spyClient = Mockito.spy(client);
Mockito.doReturn(new DummyProducerBuilder<>((PulsarClientImpl) spyClient, Schema.BYTES))
.when(spyClient).newProducer(Schema.STRING);
- // use reflection to replace the client in the broker
- Field field = PulsarService.class.getDeclaredField("client");
- field.setAccessible(true);
- field.set(pulsar, spyClient);
+ clientField.set(healthChecker, spyClient);
+
+ // change timeout to 1 second to speed up test
+ Field timeoutField = HealthChecker.class.getDeclaredField("timeout");
+ timeoutField.setAccessible(true);
+ timeoutField.set(healthChecker, Duration.ofSeconds(1));
+
try {
admin.brokers().healthcheck(TopicVersion.V2);
- throw new Exception("Should not reach here");
+ fail("Should not reach here");
} catch (PulsarAdminException e) {
log.info("Exception caught", e);
assertTrue(e.getMessage().contains("LowOverheadTimeoutException"));
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryIntegrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryIntegrationTest.java
index 189c29400c4..e975671fa12 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryIntegrationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryIntegrationTest.java
@@ -37,7 +37,7 @@ import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@Slf4j
-@Test(groups = "flaky")
+@Test(groups = "broker")
public class BrokerRegistryIntegrationTest {
private static final String clusterName = "test";
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryMetadataStoreIntegrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryMetadataStoreIntegrationTest.java
index 15097b565db..3e01b1fad0f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryMetadataStoreIntegrationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryMetadataStoreIntegrationTest.java
@@ -22,7 +22,7 @@ import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateMetadataStoreTableViewImpl;
import org.testng.annotations.Test;
-@Test(groups = "flaky")
+@Test(groups = "broker")
public class BrokerRegistryMetadataStoreIntegrationTest extends BrokerRegistryIntegrationTest {
@Override
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/systopic/PartitionedSystemTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/systopic/PartitionedSystemTopicTest.java
index e31f78665b3..fb0801181c8 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/systopic/PartitionedSystemTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/systopic/PartitionedSystemTopicTest.java
@@ -34,9 +34,9 @@ import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
import org.apache.bookkeeper.mledger.impl.NullLedgerOffloader;
import org.apache.commons.lang.RandomStringUtils;
import org.apache.commons.lang3.reflect.MethodUtils;
-import org.apache.pulsar.broker.admin.impl.BrokersBase;
import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.service.BrokerTestBase;
+import org.apache.pulsar.broker.service.HealthChecker;
import org.apache.pulsar.broker.service.TopicPolicyTestUtils;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
@@ -168,7 +168,7 @@ public class PartitionedSystemTopicTest extends BrokerTestBase {
public void testHealthCheckTopicNotOffload() throws Exception {
NamespaceName namespaceName = NamespaceService.getHeartbeatNamespaceV2(pulsar.getBrokerId(),
pulsar.getConfig());
- TopicName topicName = TopicName.get("persistent", namespaceName, BrokersBase.HEALTH_CHECK_TOPIC_SUFFIX);
+ TopicName topicName = TopicName.get("persistent", namespaceName, HealthChecker.HEALTH_CHECK_TOPIC_SUFFIX);
PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService()
.getTopic(topicName.toString(), true).get().get();
ManagedLedgerConfig config = persistentTopic.getManagedLedger().getConfig();
@@ -194,7 +194,7 @@ public class PartitionedSystemTopicTest extends BrokerTestBase {
Assert.assertTrue(optionalTopic.isEmpty());
TopicName heartbeatTopicName = TopicName.get("persistent",
- namespaceName, BrokersBase.HEALTH_CHECK_TOPIC_SUFFIX);
+ namespaceName, HealthChecker.HEALTH_CHECK_TOPIC_SUFFIX);
admin.topics().getRetention(heartbeatTopicName.toString());
optionalTopic = pulsar.getBrokerService()
.getTopic(topicName.getPartition(1).toString(), false).join();
@@ -221,7 +221,7 @@ public class PartitionedSystemTopicTest extends BrokerTestBase {
admin.brokers().healthcheck(TopicVersion.V2);
NamespaceName namespaceName = NamespaceService.getHeartbeatNamespaceV2(pulsar.getBrokerId(),
pulsar.getConfig());
- TopicName heartbeatTopicName = TopicName.get("persistent", namespaceName, BrokersBase.HEALTH_CHECK_TOPIC_SUFFIX);
+ TopicName heartbeatTopicName = TopicName.get("persistent", namespaceName, HealthChecker.HEALTH_CHECK_TOPIC_SUFFIX);
List<String> topics = getPulsar().getNamespaceService().getListOfPersistentTopics(namespaceName).join();
Assert.assertEquals(topics.size(), 1);
@@ -246,7 +246,7 @@ public class PartitionedSystemTopicTest extends BrokerTestBase {
List<String> topics = getPulsar().getNamespaceService().getListOfPersistentTopics(namespaceName).join();
Assert.assertEquals(topics.size(), 1);
TopicName heartbeatTopicName = TopicName.get("persistent",
- namespaceName, BrokersBase.HEALTH_CHECK_TOPIC_SUFFIX);
+ namespaceName, HealthChecker.HEALTH_CHECK_TOPIC_SUFFIX);
Assert.assertEquals(topics.get(0), heartbeatTopicName.toString());
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/NonStartableTestPulsarService.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/NonStartableTestPulsarService.java
index 70e386c68aa..46e41be0125 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/NonStartableTestPulsarService.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/NonStartableTestPulsarService.java
@@ -26,6 +26,7 @@ import java.io.IOException;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import org.apache.pulsar.broker.BookKeeperClientFactory;
@@ -116,13 +117,16 @@ class NonStartableTestPulsarService extends AbstractTestPulsarService {
}
@Override
- public PulsarClientImpl createClientImpl(ClientConfigurationData clientConf) throws PulsarClientException {
+ public PulsarClientImpl createClientImpl(ClientConfigurationData clientConf,
+ Consumer<PulsarClientImpl.PulsarClientImplBuilder> customizer)
+ throws PulsarClientException {
try {
return (PulsarClientImpl) getClient();
} catch (PulsarServerException e) {
throw new PulsarClientException(e);
}
}
+
@Override
protected BrokerService newBrokerService(PulsarService pulsar) throws Exception {
return getBrokerService();