You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2021/12/11 12:23:35 UTC
[pulsar] 03/10: [Broker Interceptor] Fix Pulsar didn't respond error messages when throw InterceptException (#11650)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 56fb0887e081014dc1c76cc99b4c9011478658f2
Author: GuoJiwei <te...@apache.org>
AuthorDate: Mon Aug 16 10:11:20 2021 +0800
[Broker Interceptor] Fix Pulsar didn't respond error messages when throw InterceptException (#11650)
(cherry picked from commit 4793ff7bc0aa5ea442878efc85e0bac011cba7ee)
---
.../apache/pulsar/broker/web/ExceptionHandler.java | 57 ++++++++++++++++++++
.../pulsar/broker/web/PreInterceptFilter.java | 8 +--
.../org/apache/pulsar/broker/web/WebService.java | 5 +-
.../pulsar/broker/admin/AdminApiSchemaTest.java | 4 +-
.../broker/intercept/InterceptFilterOutTest.java | 4 +-
.../pulsar/broker/web/ExceptionHandlerTest.java | 63 ++++++++++++++++++++++
.../pulsar/client/admin/internal/BaseResource.java | 2 +-
.../client/admin/internal/FunctionsImpl.java | 16 +++---
.../pulsar/client/admin/internal/SinksImpl.java | 10 ++--
.../pulsar/client/admin/internal/SourcesImpl.java | 10 ++--
.../pulsar/client/admin/internal/WorkerImpl.java | 10 ++--
.../admin/internal/http/AsyncHttpConnector.java | 16 ++++++
12 files changed, 173 insertions(+), 32 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/ExceptionHandler.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/ExceptionHandler.java
new file mode 100644
index 0000000..8b200a8
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/ExceptionHandler.java
@@ -0,0 +1,57 @@
+/**
+ * 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.web;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import javax.servlet.ServletResponse;
+import javax.servlet.http.HttpServletResponse;
+import javax.ws.rs.core.Response;
+import org.apache.pulsar.common.intercept.InterceptException;
+import org.eclipse.jetty.http.HttpVersion;
+import org.eclipse.jetty.http.MetaData;
+
+/**
+ * Exception handler for handle exception.
+ */
+public class ExceptionHandler {
+
+ public void handle(ServletResponse response, Exception ex) throws IOException {
+ if (ex instanceof InterceptException) {
+ String reason = ex.getMessage();
+ byte[] content = reason.getBytes(StandardCharsets.UTF_8);
+ MetaData.Response info = new MetaData.Response();
+ info.setHttpVersion(HttpVersion.HTTP_1_1);
+ info.setReason(reason);
+ info.setStatus(((InterceptException) ex).getErrorCode());
+ info.setContentLength(content.length);
+ if (response instanceof org.eclipse.jetty.server.Response) {
+ ((org.eclipse.jetty.server.Response) response).getHttpChannel().sendResponse(info,
+ ByteBuffer.wrap(content), true);
+ } else {
+ ((HttpServletResponse) response).sendError(((InterceptException) ex).getErrorCode(),
+ ex.getMessage());
+ }
+ } else {
+ ((HttpServletResponse) response).sendError(Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(),
+ ex.getMessage());
+ }
+ }
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PreInterceptFilter.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PreInterceptFilter.java
index 0cc9c9c..eaf1de4 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PreInterceptFilter.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PreInterceptFilter.java
@@ -30,7 +30,6 @@ import javax.servlet.ServletException;
import javax.servlet.ServletRequest;
import javax.servlet.ServletResponse;
import javax.servlet.http.HttpServletRequest;
-import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import org.apache.commons.lang3.StringUtils;
@@ -39,8 +38,11 @@ public class PreInterceptFilter implements Filter {
private final BrokerInterceptor interceptor;
- public PreInterceptFilter(BrokerInterceptor interceptor) {
+ private final ExceptionHandler exceptionHandler;
+
+ public PreInterceptFilter(BrokerInterceptor interceptor, ExceptionHandler exceptionHandler) {
this.interceptor = interceptor;
+ this.exceptionHandler = exceptionHandler;
}
@Override
@@ -67,7 +69,7 @@ public class PreInterceptFilter implements Filter {
interceptor.onWebserviceRequest(requestWrapper);
filterChain.doFilter(requestWrapper, servletResponse);
} catch (InterceptException e) {
- ((HttpServletResponse) servletResponse).sendError(e.getErrorCode(), e.getMessage());
+ exceptionHandler.handle(servletResponse, e);
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java
index 35df375..29a287e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java
@@ -157,9 +157,10 @@ public class WebService implements AutoCloseable {
}
if (!pulsar.getConfig().getBrokerInterceptors().isEmpty() || !pulsar.getConfig().isDisableBrokerInterceptors()) {
+ ExceptionHandler handler = new ExceptionHandler();
// Enable PreInterceptFilter only when interceptors are enabled
- context.addFilter(new FilterHolder(new PreInterceptFilter(pulsar.getBrokerInterceptor())),
- MATCH_ALL, EnumSet.allOf(DispatcherType.class));
+ context.addFilter(new FilterHolder(new PreInterceptFilter(pulsar.getBrokerInterceptor(), handler)),
+ MATCH_ALL, EnumSet.allOf(DispatcherType.class));
}
if (requiresAuthentication && pulsar.getConfiguration().isAuthenticationEnabled()) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaTest.java
index d4163b7..b88f936 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaTest.java
@@ -184,7 +184,7 @@ public class AdminApiSchemaTest extends MockedPulsarServiceBaseTest {
admin.schemas().createSchema(topicName, foo1SchemaInfo);
fail("Should have failed");
} catch (PulsarAdminException.ConflictException e) {
- assertTrue(e.getMessage().contains("HTTP 409 Conflict"));
+ assertTrue(e.getMessage().contains("HTTP 409"));
}
namespace = "schematest/testnotfound";
@@ -194,7 +194,7 @@ public class AdminApiSchemaTest extends MockedPulsarServiceBaseTest {
admin.schemas().createSchema(topicName, fooSchemaInfo);
fail("Should have failed");
} catch (PulsarAdminException.NotFoundException e) {
- assertTrue(e.getMessage().contains("HTTP 404 Not Found"));
+ assertTrue(e.getMessage().contains("HTTP 404"));
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/InterceptFilterOutTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/InterceptFilterOutTest.java
index fd13409..7ef26dd2 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/InterceptFilterOutTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/InterceptFilterOutTest.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.broker.intercept;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.web.ExceptionHandler;
import org.apache.pulsar.broker.web.PreInterceptFilter;
import org.apache.pulsar.broker.web.ResponseHandlerFilter;
import org.mockito.Mockito;
@@ -57,7 +58,8 @@ public class InterceptFilterOutTest {
@Test
public void testFilterOutForPreInterceptFilter() throws Exception {
CounterBrokerInterceptor interceptor = new CounterBrokerInterceptor();
- PreInterceptFilter filter = new PreInterceptFilter(interceptor);
+ ExceptionHandler handler = new ExceptionHandler();
+ PreInterceptFilter filter = new PreInterceptFilter(interceptor, handler);
HttpServletRequest request = Mockito.mock(HttpServletRequest.class);
HttpServletResponse response = Mockito.mock(HttpServletResponse.class);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/ExceptionHandlerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/ExceptionHandlerTest.java
new file mode 100644
index 0000000..ed70869
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/ExceptionHandlerTest.java
@@ -0,0 +1,63 @@
+/**
+ * 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.web;
+
+import lombok.SneakyThrows;
+import org.apache.pulsar.common.intercept.InterceptException;
+import org.eclipse.jetty.server.HttpChannel;
+import org.eclipse.jetty.server.Response;
+import org.mockito.Mockito;
+import org.testng.annotations.Test;
+
+import javax.servlet.http.HttpServletResponse;
+
+import static org.eclipse.jetty.http.HttpStatus.INTERNAL_SERVER_ERROR_500;
+import static org.eclipse.jetty.http.HttpStatus.PRECONDITION_FAILED_412;
+
+/**
+ * Unit test for ExceptionHandler.
+ */
+@Test(groups = "broker")
+public class ExceptionHandlerTest {
+
+ @Test
+ @SneakyThrows
+ public void testHandle() {
+ String restriction = "Reach the max tenants [5] restriction";
+ String internal = "internal exception";
+ String illegal = "illegal argument exception ";
+ ExceptionHandler handler = new ExceptionHandler();
+ HttpServletResponse response = Mockito.mock(HttpServletResponse.class);
+ handler.handle(response, new InterceptException(PRECONDITION_FAILED_412, restriction));
+ Mockito.verify(response).sendError(PRECONDITION_FAILED_412, restriction);
+
+ handler.handle(response, new InterceptException(INTERNAL_SERVER_ERROR_500, internal));
+ Mockito.verify(response).sendError(INTERNAL_SERVER_ERROR_500, internal);
+
+ handler.handle(response, new IllegalArgumentException(illegal));
+ Mockito.verify(response).sendError(INTERNAL_SERVER_ERROR_500, illegal);
+
+ Response response2 = Mockito.mock(Response.class);
+ HttpChannel httpChannel = Mockito.mock(HttpChannel.class);
+ Mockito.when(response2.getHttpChannel()).thenReturn(httpChannel);
+ handler.handle(response2, new InterceptException(PRECONDITION_FAILED_412, restriction));
+ Mockito.verify(httpChannel).sendResponse(Mockito.any(), Mockito.any(), Mockito.anyBoolean());
+ }
+
+}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BaseResource.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BaseResource.java
index ac63aae..1ff3042 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BaseResource.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BaseResource.java
@@ -236,7 +236,7 @@ public abstract class BaseResource {
}
public PulsarAdminException getApiException(Response response) {
- if (response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() == Response.Status.OK.getStatusCode()) {
return null;
}
try {
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java
index 0290acf..4916a60 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java
@@ -105,7 +105,7 @@ public class FunctionsImpl extends ComponentResource implements Functions {
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
List<String> functions = response.readEntity(new GenericType<List<String>>() {});
@@ -144,7 +144,7 @@ public class FunctionsImpl extends ComponentResource implements Functions {
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(FunctionConfig.class));
@@ -183,7 +183,7 @@ public class FunctionsImpl extends ComponentResource implements Functions {
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(FunctionStatus.class));
@@ -226,7 +226,7 @@ public class FunctionsImpl extends ComponentResource implements Functions {
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(
@@ -269,7 +269,7 @@ public class FunctionsImpl extends ComponentResource implements Functions {
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(
@@ -309,7 +309,7 @@ public class FunctionsImpl extends ComponentResource implements Functions {
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(FunctionStats.class));
@@ -919,7 +919,7 @@ public class FunctionsImpl extends ComponentResource implements Functions {
public List<ConnectorDefinition> getConnectorsList() throws PulsarAdminException {
try {
Response response = request(functions.path("connectors")).get();
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
throw getApiException(response);
}
return response.readEntity(new GenericType<List<ConnectorDefinition>>() {
@@ -976,7 +976,7 @@ public class FunctionsImpl extends ComponentResource implements Functions {
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(FunctionState.class));
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SinksImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SinksImpl.java
index a439823..1f5e704 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SinksImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SinksImpl.java
@@ -89,7 +89,7 @@ public class SinksImpl extends ComponentResource implements Sinks, Sink {
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(new GenericType<List<String>>() {}));
@@ -126,7 +126,7 @@ public class SinksImpl extends ComponentResource implements Sinks, Sink {
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(SinkConfig.class));
@@ -164,7 +164,7 @@ public class SinksImpl extends ComponentResource implements Sinks, Sink {
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(SinkStatus.class));
@@ -204,7 +204,7 @@ public class SinksImpl extends ComponentResource implements Sinks, Sink {
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(
@@ -584,7 +584,7 @@ public class SinksImpl extends ComponentResource implements Sinks, Sink {
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SourcesImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SourcesImpl.java
index 155bb54..1edd16f 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SourcesImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SourcesImpl.java
@@ -89,7 +89,7 @@ public class SourcesImpl extends ComponentResource implements Sources, Source {
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(new GenericType<List<String>>() {}));
@@ -126,7 +126,7 @@ public class SourcesImpl extends ComponentResource implements Sources, Source {
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(SourceConfig.class));
@@ -164,7 +164,7 @@ public class SourcesImpl extends ComponentResource implements Sources, Source {
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(SourceStatus.class));
@@ -205,7 +205,7 @@ public class SourcesImpl extends ComponentResource implements Sources, Source {
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(
@@ -588,7 +588,7 @@ public class SourcesImpl extends ComponentResource implements Sources, Source {
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(getApiException(response));
} else {
future.complete(response.readEntity(
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/WorkerImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/WorkerImpl.java
index 8fbeb0a..9e1c775 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/WorkerImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/WorkerImpl.java
@@ -75,7 +75,7 @@ public class WorkerImpl extends BaseResource implements Worker {
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(new ClientErrorException(response));
} else {
List<WorkerFunctionInstanceStats> metricsList =
@@ -114,7 +114,7 @@ public class WorkerImpl extends BaseResource implements Worker {
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(new ClientErrorException(response));
} else {
future.complete(response.readEntity(
@@ -153,7 +153,7 @@ public class WorkerImpl extends BaseResource implements Worker {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(new ClientErrorException(response));
} else {
future.complete(response.readEntity(new GenericType<List<WorkerInfo>>() {}));
@@ -190,7 +190,7 @@ public class WorkerImpl extends BaseResource implements Worker {
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(new ClientErrorException(response));
} else {
future.complete(response.readEntity(new GenericType<WorkerInfo>(){}));
@@ -227,7 +227,7 @@ public class WorkerImpl extends BaseResource implements Worker {
new InvocationCallback<Response>() {
@Override
public void completed(Response response) {
- if (!response.getStatusInfo().equals(Response.Status.OK)) {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
future.completeExceptionally(new ClientErrorException(response));
} else {
future.complete(response.readEntity(
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnector.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnector.java
index 1cb21ff..97b7937 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnector.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnector.java
@@ -205,6 +205,22 @@ public class AsyncHttpConnector implements Connector {
} else {
ClientResponse jerseyResponse =
new ClientResponse(Status.fromStatusCode(response.getStatusCode()), jerseyRequest);
+ jerseyResponse.setStatusInfo(new javax.ws.rs.core.Response.StatusType() {
+ @Override
+ public int getStatusCode() {
+ return response.getStatusCode();
+ }
+
+ @Override
+ public Status.Family getFamily() {
+ return Status.Family.familyOf(response.getStatusCode());
+ }
+
+ @Override
+ public String getReasonPhrase() {
+ return response.getStatusText();
+ }
+ });
response.getHeaders().forEach(e -> jerseyResponse.header(e.getKey(), e.getValue()));
if (response.hasResponseBody()) {
jerseyResponse.setEntityStream(response.getResponseBodyAsStream());