This is an automated email from the ASF dual-hosted git repository. guangning pushed a commit to branch branch-2.5 in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 15f876a3525c8e7e54e276cc049a2c5d98f06864 Author: Masahiro Sakamoto <[email protected]> AuthorDate: Tue Jan 28 18:37:21 2020 +0900 Add timeout to search for web service URLs to avoid web threads getting stuck (#6124) --- .../apache/pulsar/broker/namespace/NamespaceService.java | 13 ++++++++----- .../org/apache/pulsar/broker/web/PulsarWebResource.java | 9 +++++++++ 2 files changed, 17 insertions(+), 5 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java index 91eea07..0770a4d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java @@ -196,22 +196,25 @@ public class NamespaceService { * * If the service unit is not owned, return an empty optional */ - public Optional<URL> getWebServiceUrl(ServiceUnitId suName, boolean authoritative, boolean isRequestHttps, boolean readOnly) - throws Exception { + public Optional<URL> getWebServiceUrl(ServiceUnitId suName, boolean authoritative, boolean isRequestHttps, + boolean readOnly) throws Exception { if (suName instanceof TopicName) { TopicName name = (TopicName) suName; if (LOG.isDebugEnabled()) { LOG.debug("Getting web service URL of topic: {} - auth: {}", name, authoritative); } - return this.internalGetWebServiceUrl(getBundle(name), authoritative, isRequestHttps, readOnly).get(); + return this.internalGetWebServiceUrl(getBundle(name), authoritative, isRequestHttps, readOnly) + .get(pulsar.getConfiguration().getZooKeeperOperationTimeoutSeconds(), SECONDS); } if (suName instanceof NamespaceName) { - return this.internalGetWebServiceUrl(getFullBundle((NamespaceName) suName), authoritative, isRequestHttps, readOnly).get(); + return this.internalGetWebServiceUrl(getFullBundle((NamespaceName) suName), authoritative, isRequestHttps, + readOnly).get(pulsar.getConfiguration().getZooKeeperOperationTimeoutSeconds(), SECONDS); } if (suName instanceof NamespaceBundle) { - return this.internalGetWebServiceUrl((NamespaceBundle) suName, authoritative, isRequestHttps, readOnly).get(); + return this.internalGetWebServiceUrl((NamespaceBundle) suName, authoritative, isRequestHttps, readOnly) + .get(pulsar.getConfiguration().getZooKeeperOperationTimeoutSeconds(), SECONDS); } throw new IllegalArgumentException("Unrecognized class of NamespaceBundle: " + suName.getClass().getName()); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java index 166551c..941df28 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java @@ -31,6 +31,7 @@ import java.util.Optional; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeoutException; import javax.servlet.ServletContext; import javax.servlet.http.HttpServletRequest; @@ -542,6 +543,10 @@ public abstract class PulsarWebResource { log.debug("Redirecting the rest call to {}", redirect); throw new WebApplicationException(Response.temporaryRedirect(redirect).build()); } + } catch (TimeoutException te) { + String msg = String.format("Finding owner for ServiceUnit %s timed out", bundle); + log.error(msg, te); + throw new RestException(Status.INTERNAL_SERVER_ERROR, msg); } catch (IllegalArgumentException iae) { // namespace format is not valid log.debug(String.format("Failed to find owner for ServiceUnit %s", bundle), iae); @@ -590,6 +595,10 @@ public abstract class PulsarWebResource { log.debug("Redirecting the rest call to {}", redirect); throw new WebApplicationException(Response.temporaryRedirect(redirect).build()); } + } catch (TimeoutException te) { + String msg = String.format("Finding owner for topic %s timed out", topicName); + log.error(msg, te); + throw new RestException(Status.INTERNAL_SERVER_ERROR, msg); } catch (IllegalArgumentException iae) { // namespace format is not valid log.debug(String.format("Failed to find owner for topic :%s", topicName), iae);
