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 2024/08/07 10:50:31 UTC
(pulsar) branch branch-3.3 updated (3c9a2df0081 -> 80090b259a4)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 3c9a2df0081 [improve][misc] Optimize TLS performance by omitting extra buffer copies (#23115)
new fe427484246 [fix][client] Fix timeout handling in Pulsar Admin client (#23128)
new 80090b259a4 [fix][build] Remove unnecessary Oracle maven repository from pom.xml (#23132)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
pom.xml | 8 --
.../pulsar/client/admin/internal/BaseResource.java | 8 +-
.../pulsar/client/admin/internal/BookiesImpl.java | 4 +-
.../client/admin/internal/BrokerStatsImpl.java | 4 +-
.../pulsar/client/admin/internal/BrokersImpl.java | 4 +-
.../pulsar/client/admin/internal/ClustersImpl.java | 4 +-
.../client/admin/internal/ComponentResource.java | 4 +-
.../client/admin/internal/FunctionsImpl.java | 4 +-
.../client/admin/internal/NamespacesImpl.java | 4 +-
.../admin/internal/NonPersistentTopicsImpl.java | 4 +-
.../pulsar/client/admin/internal/PackagesImpl.java | 4 +-
.../client/admin/internal/ProxyStatsImpl.java | 4 +-
.../client/admin/internal/PulsarAdminImpl.java | 46 +++----
.../client/admin/internal/ResourceGroupsImpl.java | 4 +-
.../client/admin/internal/ResourceQuotasImpl.java | 4 +-
.../pulsar/client/admin/internal/SchemasImpl.java | 4 +-
.../pulsar/client/admin/internal/SinksImpl.java | 4 +-
.../pulsar/client/admin/internal/SourcesImpl.java | 4 +-
.../pulsar/client/admin/internal/TenantsImpl.java | 4 +-
.../pulsar/client/admin/internal/TopicsImpl.java | 4 +-
.../client/admin/internal/TransactionsImpl.java | 4 +-
.../pulsar/client/admin/internal/WorkerImpl.java | 4 +-
.../admin/internal/http/AsyncHttpConnector.java | 37 ++++--
.../internal/http/AsyncHttpConnectorTest.java | 140 +++++++++++++++++++++
.../src/test/resources/log4j2.xml | 1 +
.../java/org/apache/pulsar/admin/cli/CmdBase.java | 14 +--
pulsar-io/rabbitmq/pom.xml | 14 +++
.../pulsar/io/rabbitmq/RabbitMQBrokerManager.java | 18 +--
.../pulsar/io/rabbitmq/sink/RabbitMQSinkTest.java | 9 +-
.../io/rabbitmq/source/RabbitMQSourceTest.java | 10 +-
pulsar-io/rabbitmq/src/test/resources/qpid.json | 59 ++++++---
31 files changed, 315 insertions(+), 125 deletions(-)
create mode 100644 pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnectorTest.java
copy {pulsar-broker => pulsar-client-admin}/src/test/resources/log4j2.xml (96%)
(pulsar) 01/02: [fix][client] Fix timeout handling in Pulsar Admin client (#23128)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit fe4274842465f922a5bdac40294b84b1c9fddc94
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Aug 7 08:22:25 2024 +0300
[fix][client] Fix timeout handling in Pulsar Admin client (#23128)
(cherry picked from commit 3b01c96594ae1af215018b1e1df29e5416f240d9)
---
.../pulsar/client/admin/internal/BaseResource.java | 8 +-
.../pulsar/client/admin/internal/BookiesImpl.java | 4 +-
.../client/admin/internal/BrokerStatsImpl.java | 4 +-
.../pulsar/client/admin/internal/BrokersImpl.java | 4 +-
.../pulsar/client/admin/internal/ClustersImpl.java | 4 +-
.../client/admin/internal/ComponentResource.java | 4 +-
.../client/admin/internal/FunctionsImpl.java | 4 +-
.../client/admin/internal/NamespacesImpl.java | 4 +-
.../admin/internal/NonPersistentTopicsImpl.java | 4 +-
.../pulsar/client/admin/internal/PackagesImpl.java | 4 +-
.../client/admin/internal/ProxyStatsImpl.java | 4 +-
.../client/admin/internal/PulsarAdminImpl.java | 46 +++----
.../client/admin/internal/ResourceGroupsImpl.java | 4 +-
.../client/admin/internal/ResourceQuotasImpl.java | 4 +-
.../pulsar/client/admin/internal/SchemasImpl.java | 4 +-
.../pulsar/client/admin/internal/SinksImpl.java | 4 +-
.../pulsar/client/admin/internal/SourcesImpl.java | 4 +-
.../pulsar/client/admin/internal/TenantsImpl.java | 4 +-
.../pulsar/client/admin/internal/TopicsImpl.java | 4 +-
.../client/admin/internal/TransactionsImpl.java | 4 +-
.../pulsar/client/admin/internal/WorkerImpl.java | 4 +-
.../admin/internal/http/AsyncHttpConnector.java | 37 ++++--
.../internal/http/AsyncHttpConnectorTest.java | 140 +++++++++++++++++++++
pulsar-client-admin/src/test/resources/log4j2.xml | 41 ++++++
.../java/org/apache/pulsar/admin/cli/CmdBase.java | 14 +--
25 files changed, 280 insertions(+), 82 deletions(-)
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 22550666cb6..ea39053c2ce 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
@@ -62,11 +62,11 @@ public abstract class BaseResource {
private static final Logger log = LoggerFactory.getLogger(BaseResource.class);
protected final Authentication auth;
- protected final long readTimeoutMs;
+ protected final long requestTimeoutMs;
- protected BaseResource(Authentication auth, long readTimeoutMs) {
+ protected BaseResource(Authentication auth, long requestTimeoutMs) {
this.auth = auth;
- this.readTimeoutMs = readTimeoutMs;
+ this.requestTimeoutMs = requestTimeoutMs;
}
public Builder request(final WebTarget target) throws PulsarAdminException {
@@ -339,7 +339,7 @@ public abstract class BaseResource {
protected <T> T sync(Supplier<CompletableFuture<T>> executor) throws PulsarAdminException {
try {
- return executor.get().get(this.readTimeoutMs, TimeUnit.MILLISECONDS);
+ return executor.get().get(this.requestTimeoutMs, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new PulsarAdminException(e);
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BookiesImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BookiesImpl.java
index 2286fb8c8a3..0bf92e02677 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BookiesImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BookiesImpl.java
@@ -32,8 +32,8 @@ import org.apache.pulsar.common.policies.data.BookiesRackConfiguration;
public class BookiesImpl extends BaseResource implements Bookies {
private final WebTarget adminBookies;
- public BookiesImpl(WebTarget web, Authentication auth, long readTimeoutMs) {
- super(auth, readTimeoutMs);
+ public BookiesImpl(WebTarget web, Authentication auth, long requestTimeoutMs) {
+ super(auth, requestTimeoutMs);
adminBookies = web.path("/admin/v2/bookies");
}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokerStatsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokerStatsImpl.java
index e409d6f4492..6ddabe9837e 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokerStatsImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokerStatsImpl.java
@@ -38,8 +38,8 @@ public class BrokerStatsImpl extends BaseResource implements BrokerStats {
private final WebTarget adminBrokerStats;
private final WebTarget adminV2BrokerStats;
- public BrokerStatsImpl(WebTarget target, Authentication auth, long readTimeoutMs) {
- super(auth, readTimeoutMs);
+ public BrokerStatsImpl(WebTarget target, Authentication auth, long requestTimeoutMs) {
+ super(auth, requestTimeoutMs);
adminBrokerStats = target.path("/admin/broker-stats");
adminV2BrokerStats = target.path("/admin/v2/broker-stats");
}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokersImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokersImpl.java
index 7b4ebb1778d..b82c3fd0f41 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokersImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokersImpl.java
@@ -37,8 +37,8 @@ import org.apache.pulsar.common.util.Codec;
public class BrokersImpl extends BaseResource implements Brokers {
private final WebTarget adminBrokers;
- public BrokersImpl(WebTarget web, Authentication auth, long readTimeoutMs) {
- super(auth, readTimeoutMs);
+ public BrokersImpl(WebTarget web, Authentication auth, long requestTimeoutMs) {
+ super(auth, requestTimeoutMs);
adminBrokers = web.path("admin/v2/brokers");
}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ClustersImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ClustersImpl.java
index 231d4506d61..24048ea3c0a 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ClustersImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ClustersImpl.java
@@ -47,8 +47,8 @@ public class ClustersImpl extends BaseResource implements Clusters {
private final WebTarget adminClusters;
- public ClustersImpl(WebTarget web, Authentication auth, long readTimeoutMs) {
- super(auth, readTimeoutMs);
+ public ClustersImpl(WebTarget web, Authentication auth, long requestTimeoutMs) {
+ super(auth, requestTimeoutMs);
adminClusters = web.path("/admin/v2/clusters");
}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ComponentResource.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ComponentResource.java
index 8beecff3897..0301f0fc2ee 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ComponentResource.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ComponentResource.java
@@ -37,8 +37,8 @@ import org.asynchttpclient.RequestBuilder;
*/
public class ComponentResource extends BaseResource {
- protected ComponentResource(Authentication auth, long readTimeoutMs) {
- super(auth, readTimeoutMs);
+ protected ComponentResource(Authentication auth, long requestTimeoutMs) {
+ super(auth, requestTimeoutMs);
}
public RequestBuilder addAuthHeaders(WebTarget target, RequestBuilder requestBuilder) throws PulsarAdminException {
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 bb4cb0c1ef8..97c42e5c1a9 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
@@ -72,8 +72,8 @@ public class FunctionsImpl extends ComponentResource implements Functions {
private final WebTarget functions;
private final AsyncHttpClient asyncHttpClient;
- public FunctionsImpl(WebTarget web, Authentication auth, AsyncHttpClient asyncHttpClient, long readTimeoutMs) {
- super(auth, readTimeoutMs);
+ public FunctionsImpl(WebTarget web, Authentication auth, AsyncHttpClient asyncHttpClient, long requestTimeoutMs) {
+ super(auth, requestTimeoutMs);
this.functions = web.path("/admin/v3/functions");
this.asyncHttpClient = asyncHttpClient;
}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NamespacesImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NamespacesImpl.java
index c7492a26ab3..7d41c7203d2 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NamespacesImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NamespacesImpl.java
@@ -64,8 +64,8 @@ public class NamespacesImpl extends BaseResource implements Namespaces {
private final WebTarget adminNamespaces;
private final WebTarget adminV2Namespaces;
- public NamespacesImpl(WebTarget web, Authentication auth, long readTimeoutMs) {
- super(auth, readTimeoutMs);
+ public NamespacesImpl(WebTarget web, Authentication auth, long requestTimeoutMs) {
+ super(auth, requestTimeoutMs);
adminNamespaces = web.path("/admin/namespaces");
adminV2Namespaces = web.path("/admin/v2/namespaces");
}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NonPersistentTopicsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NonPersistentTopicsImpl.java
index 76727cd1e0f..e98d44fdc4a 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NonPersistentTopicsImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NonPersistentTopicsImpl.java
@@ -38,8 +38,8 @@ public class NonPersistentTopicsImpl extends BaseResource implements NonPersiste
private final WebTarget adminNonPersistentTopics;
private final WebTarget adminV2NonPersistentTopics;
- public NonPersistentTopicsImpl(WebTarget web, Authentication auth, long readTimeoutMs) {
- super(auth, readTimeoutMs);
+ public NonPersistentTopicsImpl(WebTarget web, Authentication auth, long requestTimeoutMs) {
+ super(auth, requestTimeoutMs);
adminNonPersistentTopics = web.path("/admin");
adminV2NonPersistentTopics = web.path("/admin/v2");
}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PackagesImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PackagesImpl.java
index 694c2160b0f..d69bef448c1 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PackagesImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PackagesImpl.java
@@ -57,8 +57,8 @@ public class PackagesImpl extends ComponentResource implements Packages {
private final WebTarget packages;
private final AsyncHttpClient httpClient;
- public PackagesImpl(WebTarget webTarget, Authentication auth, AsyncHttpClient client, long readTimeoutMs) {
- super(auth, readTimeoutMs);
+ public PackagesImpl(WebTarget webTarget, Authentication auth, AsyncHttpClient client, long requestTimeoutMs) {
+ super(auth, requestTimeoutMs);
this.httpClient = client;
this.packages = webTarget.path("/admin/v3/packages");
}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ProxyStatsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ProxyStatsImpl.java
index e98d9bf57b3..7ed07a1a6ad 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ProxyStatsImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ProxyStatsImpl.java
@@ -32,8 +32,8 @@ public class ProxyStatsImpl extends BaseResource implements ProxyStats {
private final WebTarget adminProxyStats;
- public ProxyStatsImpl(WebTarget target, Authentication auth, long readTimeoutMs) {
- super(auth, readTimeoutMs);
+ public ProxyStatsImpl(WebTarget target, Authentication auth, long requestTimeoutMs) {
+ super(auth, requestTimeoutMs);
adminProxyStats = target.path("/proxy-stats");
}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java
index 39347850cf6..e00caa6dbbc 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java
@@ -159,29 +159,29 @@ public class PulsarAdminImpl implements PulsarAdmin {
Math.toIntExact(clientConfigData.getRequestTimeoutMs()),
clientConfigData.getAutoCertRefreshSeconds());
- long readTimeoutMs = clientConfigData.getReadTimeoutMs();
- this.clusters = new ClustersImpl(root, auth, readTimeoutMs);
- this.brokers = new BrokersImpl(root, auth, readTimeoutMs);
- this.brokerStats = new BrokerStatsImpl(root, auth, readTimeoutMs);
- this.proxyStats = new ProxyStatsImpl(root, auth, readTimeoutMs);
- this.tenants = new TenantsImpl(root, auth, readTimeoutMs);
- this.resourcegroups = new ResourceGroupsImpl(root, auth, readTimeoutMs);
- this.properties = new TenantsImpl(root, auth, readTimeoutMs);
- this.namespaces = new NamespacesImpl(root, auth, readTimeoutMs);
- this.topics = new TopicsImpl(root, auth, readTimeoutMs);
- this.localTopicPolicies = new TopicPoliciesImpl(root, auth, readTimeoutMs, false);
- this.globalTopicPolicies = new TopicPoliciesImpl(root, auth, readTimeoutMs, true);
- this.nonPersistentTopics = new NonPersistentTopicsImpl(root, auth, readTimeoutMs);
- this.resourceQuotas = new ResourceQuotasImpl(root, auth, readTimeoutMs);
- this.lookups = new LookupImpl(root, auth, useTls, readTimeoutMs, topics);
- this.functions = new FunctionsImpl(root, auth, asyncHttpConnector.getHttpClient(), readTimeoutMs);
- this.sources = new SourcesImpl(root, auth, asyncHttpConnector.getHttpClient(), readTimeoutMs);
- this.sinks = new SinksImpl(root, auth, asyncHttpConnector.getHttpClient(), readTimeoutMs);
- this.worker = new WorkerImpl(root, auth, readTimeoutMs);
- this.schemas = new SchemasImpl(root, auth, readTimeoutMs);
- this.bookies = new BookiesImpl(root, auth, readTimeoutMs);
- this.packages = new PackagesImpl(root, auth, asyncHttpConnector.getHttpClient(), readTimeoutMs);
- this.transactions = new TransactionsImpl(root, auth, readTimeoutMs);
+ long requestTimeoutMs = clientConfigData.getRequestTimeoutMs();
+ this.clusters = new ClustersImpl(root, auth, requestTimeoutMs);
+ this.brokers = new BrokersImpl(root, auth, requestTimeoutMs);
+ this.brokerStats = new BrokerStatsImpl(root, auth, requestTimeoutMs);
+ this.proxyStats = new ProxyStatsImpl(root, auth, requestTimeoutMs);
+ this.tenants = new TenantsImpl(root, auth, requestTimeoutMs);
+ this.resourcegroups = new ResourceGroupsImpl(root, auth, requestTimeoutMs);
+ this.properties = new TenantsImpl(root, auth, requestTimeoutMs);
+ this.namespaces = new NamespacesImpl(root, auth, requestTimeoutMs);
+ this.topics = new TopicsImpl(root, auth, requestTimeoutMs);
+ this.localTopicPolicies = new TopicPoliciesImpl(root, auth, requestTimeoutMs, false);
+ this.globalTopicPolicies = new TopicPoliciesImpl(root, auth, requestTimeoutMs, true);
+ this.nonPersistentTopics = new NonPersistentTopicsImpl(root, auth, requestTimeoutMs);
+ this.resourceQuotas = new ResourceQuotasImpl(root, auth, requestTimeoutMs);
+ this.lookups = new LookupImpl(root, auth, useTls, requestTimeoutMs, topics);
+ this.functions = new FunctionsImpl(root, auth, asyncHttpConnector.getHttpClient(), requestTimeoutMs);
+ this.sources = new SourcesImpl(root, auth, asyncHttpConnector.getHttpClient(), requestTimeoutMs);
+ this.sinks = new SinksImpl(root, auth, asyncHttpConnector.getHttpClient(), requestTimeoutMs);
+ this.worker = new WorkerImpl(root, auth, requestTimeoutMs);
+ this.schemas = new SchemasImpl(root, auth, requestTimeoutMs);
+ this.bookies = new BookiesImpl(root, auth, requestTimeoutMs);
+ this.packages = new PackagesImpl(root, auth, asyncHttpConnector.getHttpClient(), requestTimeoutMs);
+ this.transactions = new TransactionsImpl(root, auth, requestTimeoutMs);
if (originalCtxLoader != null) {
Thread.currentThread().setContextClassLoader(originalCtxLoader);
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ResourceGroupsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ResourceGroupsImpl.java
index a8cef60232f..4e7230eebd9 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ResourceGroupsImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ResourceGroupsImpl.java
@@ -32,8 +32,8 @@ import org.apache.pulsar.common.policies.data.ResourceGroup;
public class ResourceGroupsImpl extends BaseResource implements ResourceGroups {
private final WebTarget adminResourceGroups;
- public ResourceGroupsImpl(WebTarget web, Authentication auth, long readTimeoutMs) {
- super(auth, readTimeoutMs);
+ public ResourceGroupsImpl(WebTarget web, Authentication auth, long requestTimeoutMs) {
+ super(auth, requestTimeoutMs);
adminResourceGroups = web.path("/admin/v2/resourcegroups");
}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ResourceQuotasImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ResourceQuotasImpl.java
index 1e80c9eda94..68884d99448 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ResourceQuotasImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ResourceQuotasImpl.java
@@ -33,8 +33,8 @@ public class ResourceQuotasImpl extends BaseResource implements ResourceQuotas {
private final WebTarget adminQuotas;
private final WebTarget adminV2Quotas;
- public ResourceQuotasImpl(WebTarget web, Authentication auth, long readTimeoutMs) {
- super(auth, readTimeoutMs);
+ public ResourceQuotasImpl(WebTarget web, Authentication auth, long requestTimeoutMs) {
+ super(auth, requestTimeoutMs);
adminQuotas = web.path("/admin/resource-quotas");
adminV2Quotas = web.path("/admin/v2/resource-quotas");
}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SchemasImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SchemasImpl.java
index 593eb67fc0d..28b435ab567 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SchemasImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SchemasImpl.java
@@ -46,8 +46,8 @@ public class SchemasImpl extends BaseResource implements Schemas {
private final WebTarget adminV2;
private final WebTarget adminV1;
- public SchemasImpl(WebTarget web, Authentication auth, long readTimeoutMs) {
- super(auth, readTimeoutMs);
+ public SchemasImpl(WebTarget web, Authentication auth, long requestTimeoutMs) {
+ super(auth, requestTimeoutMs);
this.adminV1 = web.path("/admin/schemas");
this.adminV2 = web.path("/admin/v2/schemas");
}
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 c14f75ab367..a30f51264cc 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
@@ -53,8 +53,8 @@ public class SinksImpl extends ComponentResource implements Sinks, Sink {
private final WebTarget sink;
private final AsyncHttpClient asyncHttpClient;
- public SinksImpl(WebTarget web, Authentication auth, AsyncHttpClient asyncHttpClient, long readTimeoutMs) {
- super(auth, readTimeoutMs);
+ public SinksImpl(WebTarget web, Authentication auth, AsyncHttpClient asyncHttpClient, long requestTimeoutMs) {
+ super(auth, requestTimeoutMs);
this.sink = web.path("/admin/v3/sink");
this.asyncHttpClient = asyncHttpClient;
}
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 6e5b84c7f04..8821ed61ce5 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
@@ -52,8 +52,8 @@ public class SourcesImpl extends ComponentResource implements Sources, Source {
private final WebTarget source;
private final AsyncHttpClient asyncHttpClient;
- public SourcesImpl(WebTarget web, Authentication auth, AsyncHttpClient asyncHttpClient, long readTimeoutMs) {
- super(auth, readTimeoutMs);
+ public SourcesImpl(WebTarget web, Authentication auth, AsyncHttpClient asyncHttpClient, long requestTimeoutMs) {
+ super(auth, requestTimeoutMs);
this.source = web.path("/admin/v3/source");
this.asyncHttpClient = asyncHttpClient;
}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TenantsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TenantsImpl.java
index 9b70e39ec49..c12f3754b4a 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TenantsImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TenantsImpl.java
@@ -34,8 +34,8 @@ import org.apache.pulsar.common.policies.data.TenantInfoImpl;
public class TenantsImpl extends BaseResource implements Tenants, Properties {
private final WebTarget adminTenants;
- public TenantsImpl(WebTarget web, Authentication auth, long readTimeoutMs) {
- super(auth, readTimeoutMs);
+ public TenantsImpl(WebTarget web, Authentication auth, long requestTimeoutMs) {
+ super(auth, requestTimeoutMs);
adminTenants = web.path("/admin/v2/tenants");
}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java
index b7a8b876640..9c4a6eef753 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java
@@ -137,8 +137,8 @@ public class TopicsImpl extends BaseResource implements Topics {
public static final String PROPERTY_SHADOW_SOURCE_KEY = "PULSAR.SHADOW_SOURCE";
- public TopicsImpl(WebTarget web, Authentication auth, long readTimeoutMs) {
- super(auth, readTimeoutMs);
+ public TopicsImpl(WebTarget web, Authentication auth, long requestTimeoutMs) {
+ super(auth, requestTimeoutMs);
adminTopics = web.path("/admin");
adminV2Topics = web.path("/admin/v2");
}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TransactionsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TransactionsImpl.java
index 460478787eb..a0b9dd234d9 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TransactionsImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TransactionsImpl.java
@@ -46,8 +46,8 @@ import org.apache.pulsar.common.stats.PositionInPendingAckStats;
public class TransactionsImpl extends BaseResource implements Transactions {
private final WebTarget adminV3Transactions;
- public TransactionsImpl(WebTarget web, Authentication auth, long readTimeoutMs) {
- super(auth, readTimeoutMs);
+ public TransactionsImpl(WebTarget web, Authentication auth, long requestTimeoutMs) {
+ super(auth, requestTimeoutMs);
adminV3Transactions = web.path("/admin/v3/transactions");
}
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 60b1226d581..12a691edb08 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
@@ -40,8 +40,8 @@ public class WorkerImpl extends BaseResource implements Worker {
private final WebTarget workerStats;
private final WebTarget worker;
- public WorkerImpl(WebTarget web, Authentication auth, long readTimeoutMs) {
- super(auth, readTimeoutMs);
+ public WorkerImpl(WebTarget web, Authentication auth, long requestTimeoutMs) {
+ super(auth, requestTimeoutMs);
this.worker = web.path("/admin/v2/worker");
this.workerStats = web.path("/admin/v2/worker-stats");
}
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 9ad0ce5029c..a0569c391ad 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
@@ -35,7 +35,6 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeoutException;
-import java.util.function.Function;
import java.util.function.Supplier;
import javax.net.ssl.SSLContext;
import javax.ws.rs.client.Client;
@@ -59,6 +58,7 @@ import org.asynchttpclient.AsyncHttpClient;
import org.asynchttpclient.BoundRequestBuilder;
import org.asynchttpclient.DefaultAsyncHttpClient;
import org.asynchttpclient.DefaultAsyncHttpClientConfig;
+import org.asynchttpclient.ListenableFuture;
import org.asynchttpclient.Request;
import org.asynchttpclient.Response;
import org.asynchttpclient.channel.DefaultKeepAliveStrategy;
@@ -74,11 +74,11 @@ import org.glassfish.jersey.client.spi.Connector;
*/
@Slf4j
public class AsyncHttpConnector implements Connector {
- private static final TimeoutException READ_TIMEOUT_EXCEPTION =
- FutureUtil.createTimeoutException("Read timeout", AsyncHttpConnector.class, "retryOrTimeout(...)");
+ private static final TimeoutException REQUEST_TIMEOUT_EXCEPTION =
+ FutureUtil.createTimeoutException("Request timeout", AsyncHttpConnector.class, "retryOrTimeout(...)");
@Getter
private final AsyncHttpClient httpClient;
- private final Duration readTimeout;
+ private final Duration requestTimeout;
private final int maxRetries;
private final PulsarServiceNameResolver serviceNameResolver;
private final ScheduledExecutorService delayer = Executors.newScheduledThreadPool(1,
@@ -185,7 +185,7 @@ public class AsyncHttpConnector implements Connector {
confBuilder.setDisableHttpsEndpointIdentificationAlgorithm(!conf.isTlsHostnameVerificationEnable());
}
httpClient = new DefaultAsyncHttpClient(confBuilder.build());
- this.readTimeout = Duration.ofMillis(readTimeoutMs);
+ this.requestTimeout = requestTimeoutMs > 0 ? Duration.ofMillis(requestTimeoutMs) : null;
this.maxRetries = httpClient.getConfig().getMaxRequestRetry();
}
@@ -264,9 +264,10 @@ public class AsyncHttpConnector implements Connector {
private CompletableFuture<Response> retryOrTimeOut(ClientRequest request) {
final CompletableFuture<Response> resultFuture = new CompletableFuture<>();
retryOperation(resultFuture, () -> oneShot(serviceNameResolver.resolveHost(), request), maxRetries);
- CompletableFuture<Response> timeoutAfter = FutureUtil.createFutureWithTimeout(readTimeout, delayer,
- () -> READ_TIMEOUT_EXCEPTION);
- return resultFuture.applyToEither(timeoutAfter, Function.identity());
+ if (requestTimeout != null) {
+ FutureUtil.addTimeoutHandling(resultFuture, requestTimeout, delayer, () -> REQUEST_TIMEOUT_EXCEPTION);
+ }
+ return resultFuture;
}
private <T> void retryOperation(
@@ -285,11 +286,18 @@ public class AsyncHttpConnector implements Connector {
new RetryException("Operation future was cancelled.", throwable));
} else {
if (retries > 0) {
+ if (log.isDebugEnabled()) {
+ log.debug("Retrying operation. Remaining retries: {}", retries);
+ }
retryOperation(
resultFuture,
operation,
retries - 1);
} else {
+ if (log.isDebugEnabled()) {
+ log.debug("Number of retries has been exhausted. Failing the operation.",
+ throwable);
+ }
resultFuture.completeExceptionally(
new RetryException("Could not complete the operation. Number of retries "
+ "has been exhausted. Failed reason: " + throwable.getMessage(),
@@ -315,7 +323,7 @@ public class AsyncHttpConnector implements Connector {
}
}
- private CompletableFuture<Response> oneShot(InetSocketAddress host, ClientRequest request) {
+ protected CompletableFuture<Response> oneShot(InetSocketAddress host, ClientRequest request) {
ClientRequest currentRequest = new ClientRequest(request);
URI newUri = replaceWithNew(host, currentRequest.getUri());
currentRequest.setUri(newUri);
@@ -347,7 +355,16 @@ public class AsyncHttpConnector implements Connector {
builder.setHeader(HttpHeaders.ACCEPT_ENCODING, "gzip");
}
- return builder.execute().toCompletableFuture();
+ ListenableFuture<Response> responseFuture = builder.execute();
+ CompletableFuture<Response> completableFuture = responseFuture.toCompletableFuture();
+ completableFuture.whenComplete((response, throwable) -> {
+ if (throwable != null && (throwable instanceof CancellationException
+ || throwable instanceof TimeoutException)) {
+ // abort the request if the future is cancelled or timed out
+ responseFuture.abort(throwable);
+ }
+ });
+ return completableFuture;
}
@Override
diff --git a/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnectorTest.java b/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnectorTest.java
new file mode 100644
index 00000000000..dd3fb40ae9a
--- /dev/null
+++ b/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnectorTest.java
@@ -0,0 +1,140 @@
+/*
+ * 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.client.admin.internal.http;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
+import static com.github.tomakehurst.wiremock.client.WireMock.get;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import com.github.tomakehurst.wiremock.WireMockServer;
+import com.github.tomakehurst.wiremock.core.WireMockConfiguration;
+import com.github.tomakehurst.wiremock.stubbing.Scenario;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.asynchttpclient.Response;
+import org.glassfish.jersey.client.ClientConfig;
+import org.glassfish.jersey.client.ClientRequest;
+import org.glassfish.jersey.client.ClientResponse;
+import org.glassfish.jersey.client.JerseyClient;
+import org.glassfish.jersey.client.JerseyClientBuilder;
+import org.glassfish.jersey.client.spi.AsyncConnectorCallback;
+import org.glassfish.jersey.internal.MapPropertiesDelegate;
+import org.glassfish.jersey.internal.PropertiesDelegate;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+public class AsyncHttpConnectorTest {
+ WireMockServer server;
+
+ @BeforeClass(alwaysRun = true)
+ void beforeClass() throws IOException {
+ server = new WireMockServer(WireMockConfiguration.wireMockConfig()
+ .port(0));
+ server.start();
+ }
+
+ @AfterClass(alwaysRun = true)
+ void afterClass() {
+ if (server != null) {
+ server.stop();
+ }
+ }
+
+ static class TestClientRequest extends ClientRequest {
+ public TestClientRequest(URI uri, ClientConfig clientConfig, PropertiesDelegate propertiesDelegate) {
+ super(uri, clientConfig, propertiesDelegate);
+ }
+ }
+
+ @Test
+ public void testShouldStopRetriesWhenTimeoutOccurs() throws IOException, ExecutionException, InterruptedException {
+ server.stubFor(get(urlEqualTo("/admin/v2/clusters"))
+ .inScenario("once")
+ .whenScenarioStateIs(Scenario.STARTED)
+ .willSetStateTo("next")
+ .willReturn(aResponse()
+ .withHeader("Content-Type", "application/json")
+ .withBody("[\"test-cluster\"]")));
+
+ server.stubFor(get(urlEqualTo("/admin/v2/clusters"))
+ .inScenario("once")
+ .whenScenarioStateIs("next")
+ .willSetStateTo("retried")
+ .willReturn(aResponse().withStatus(500)));
+
+ ClientConfigurationData conf = new ClientConfigurationData();
+ conf.setServiceUrl("http://localhost:" + server.port());
+
+ int requestTimeout = 500;
+
+ @Cleanup("shutdownNow")
+ ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor();
+ Executor delayedExecutor = runnable -> {
+ scheduledExecutor.schedule(runnable, requestTimeout, TimeUnit.MILLISECONDS);
+ };
+ @Cleanup
+ AsyncHttpConnector connector = new AsyncHttpConnector(5000, requestTimeout,
+ requestTimeout, 0, conf, false) {
+ @Override
+ protected CompletableFuture<Response> oneShot(InetSocketAddress host, ClientRequest request) {
+ // delay the response to simulate a timeout
+ return super.oneShot(host, request)
+ .thenApplyAsync(response -> {
+ return response;
+ }, delayedExecutor);
+ }
+ };
+
+ JerseyClient jerseyClient = JerseyClientBuilder.createClient();
+ ClientConfig clientConfig = jerseyClient.getConfiguration();
+ PropertiesDelegate propertiesDelegate = new MapPropertiesDelegate();
+ URI requestUri = URI.create("http://localhost:" + server.port() + "/admin/v2/clusters");
+ ClientRequest request = new TestClientRequest(requestUri, clientConfig, propertiesDelegate);
+ request.setMethod("GET");
+ CompletableFuture<ClientResponse> future = new CompletableFuture<>();
+ connector.apply(request, new AsyncConnectorCallback() {
+ @Override
+ public void response(ClientResponse response) {
+ future.complete(response);
+ }
+
+ @Override
+ public void failure(Throwable failure) {
+ future.completeExceptionally(failure);
+ }
+ });
+ Thread.sleep(2 * requestTimeout);
+ String scenarioState =
+ server.getAllScenarios().getScenarios().stream().filter(scenario -> "once".equals(scenario.getName()))
+ .findFirst().get().getState();
+ assertEquals(scenarioState, "next");
+ assertTrue(future.isCompletedExceptionally());
+ }
+}
\ No newline at end of file
diff --git a/pulsar-client-admin/src/test/resources/log4j2.xml b/pulsar-client-admin/src/test/resources/log4j2.xml
new file mode 100644
index 00000000000..9b57b450ffa
--- /dev/null
+++ b/pulsar-client-admin/src/test/resources/log4j2.xml
@@ -0,0 +1,41 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<Configuration xmlns="http://logging.apache.org/log4j/2.0/config"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://logging.apache.org/log4j/2.0/config https://logging.apache.org/log4j/2.0/log4j-core.xsd">
+ <Appenders>
+ <!-- setting follow="true" is required for using ConsoleCaptor to validate log messages -->
+ <Console name="CONSOLE" target="SYSTEM_OUT" follow="true">
+ <PatternLayout pattern="%d{ISO8601} - %-5p - [%t:%c{1}] - %m%n"/>
+ </Console>
+ </Appenders>
+ <Loggers>
+<!-- <Logger name="org.apache.pulsar.broker.service.persistent.PersistentTopic" level="DEBUG" additivity="false">-->
+<!-- <AppenderRef ref="CONSOLE" />-->
+<!-- </Logger>-->
+
+ <Root level="INFO">
+ <AppenderRef ref="CONSOLE"/>
+ </Root>
+ <Logger name="org.apache.pulsar.client.admin" level="DEBUG" />
+ </Loggers>
+</Configuration>
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdBase.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdBase.java
index 07e8a8b5df6..8ff7f1c31ce 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdBase.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdBase.java
@@ -37,10 +37,10 @@ public abstract class CmdBase {
private final Supplier<PulsarAdmin> adminSupplier;
/**
- * Default read timeout in milliseconds.
- * Used if not found from configuration data in {@link #getReadTimeoutMs()}
+ * Default request timeout in milliseconds.
+ * Used if not found from configuration data in {@link #getRequestTimeoutMs()}
*/
- private static final long DEFAULT_READ_TIMEOUT_MILLIS = 60000;
+ private static final long DEFAULT_REQUEST_TIMEOUT_MILLIS = 60000;
public CmdBase(String cmdName, Supplier<PulsarAdmin> adminSupplier) {
this.adminSupplier = adminSupplier;
@@ -56,17 +56,17 @@ public abstract class CmdBase {
return adminSupplier.get();
}
- protected long getReadTimeoutMs() {
+ protected long getRequestTimeoutMs() {
PulsarAdmin pulsarAdmin = getAdmin();
if (pulsarAdmin instanceof PulsarAdminImpl) {
- return ((PulsarAdminImpl) pulsarAdmin).getClientConfigData().getReadTimeoutMs();
+ return ((PulsarAdminImpl) pulsarAdmin).getClientConfigData().getRequestTimeoutMs();
}
- return DEFAULT_READ_TIMEOUT_MILLIS;
+ return DEFAULT_REQUEST_TIMEOUT_MILLIS;
}
protected <T> T sync(Supplier<CompletableFuture<T>> executor) throws PulsarAdminException {
try {
- return executor.get().get(getReadTimeoutMs(), TimeUnit.MILLISECONDS);
+ return executor.get().get(getRequestTimeoutMs(), TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new PulsarAdminException(e);
(pulsar) 02/02: [fix][build] Remove unnecessary Oracle maven repository from pom.xml (#23132)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 80090b259a48435203bbd92fd27811bdcb4618e0
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Aug 7 13:46:38 2024 +0300
[fix][build] Remove unnecessary Oracle maven repository from pom.xml (#23132)
(cherry picked from commit 175ea005747ea7e229b2f492b039cf51007421ad)
---
pom.xml | 8 ---
pulsar-io/rabbitmq/pom.xml | 14 +++++
.../pulsar/io/rabbitmq/RabbitMQBrokerManager.java | 18 +------
.../pulsar/io/rabbitmq/sink/RabbitMQSinkTest.java | 9 +++-
.../io/rabbitmq/source/RabbitMQSourceTest.java | 10 +++-
pulsar-io/rabbitmq/src/test/resources/qpid.json | 59 +++++++++++++++++-----
6 files changed, 75 insertions(+), 43 deletions(-)
diff --git a/pom.xml b/pom.xml
index 768e82863d9..6f3a4c44ed8 100644
--- a/pom.xml
+++ b/pom.xml
@@ -2794,13 +2794,5 @@ flexible messaging model and an intuitive client API.</description>
<enabled>false</enabled>
</snapshots>
</repository>
- <!-- For the BDB JE dependency -->
- <repository>
- <id>oracle.releases</id>
- <url>https://download.oracle.com/maven</url>
- <snapshots>
- <enabled>false</enabled>
- </snapshots>
- </repository>
</repositories>
</project>
diff --git a/pulsar-io/rabbitmq/pom.xml b/pulsar-io/rabbitmq/pom.xml
index 8b0aab5f747..fff4cdb4194 100644
--- a/pulsar-io/rabbitmq/pom.xml
+++ b/pulsar-io/rabbitmq/pom.xml
@@ -85,6 +85,20 @@
<artifactId>qpid-broker</artifactId>
<version>9.2.0</version>
<scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.qpid</groupId>
+ <artifactId>qpid-bdbstore</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.qpid</groupId>
+ <artifactId>qpid-broker-plugins-amqp-1-0-protocol-bdb-link-store</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.qpid</groupId>
+ <artifactId>qpid-broker-plugins-derby-store</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
<groupId>org.awaitility</groupId>
diff --git a/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/RabbitMQBrokerManager.java b/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/RabbitMQBrokerManager.java
index 83331bf7de8..4ff8c61e4f4 100644
--- a/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/RabbitMQBrokerManager.java
+++ b/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/RabbitMQBrokerManager.java
@@ -18,8 +18,6 @@
*/
package org.apache.pulsar.io.rabbitmq;
-import java.io.File;
-import java.io.FileOutputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.HashMap;
@@ -42,28 +40,14 @@ public class RabbitMQBrokerManager {
Map<String, Object> getBrokerOptions(String port) throws Exception {
Path tmpFolder = Files.createTempDirectory("qpidWork");
- Path homeFolder = Files.createTempDirectory("qpidHome");
- File etc = new File(homeFolder.toFile(), "etc");
- etc.mkdir();
- FileOutputStream fos = new FileOutputStream(new File(etc, "passwd"));
- fos.write("guest:guest\n".getBytes());
- fos.close();
-
Map<String, Object> config = new HashMap<>();
config.put("qpid.work_dir", tmpFolder.toAbsolutePath().toString());
config.put("qpid.amqp_port", port);
- config.put("qpid.home_dir", homeFolder.toAbsolutePath().toString());
- String configPath = getFile("qpid.json").getAbsolutePath();
Map<String, Object> context = new HashMap<>();
- context.put(SystemConfig.INITIAL_CONFIGURATION_LOCATION, configPath);
+ context.put(SystemConfig.INITIAL_CONFIGURATION_LOCATION, "classpath:qpid.json");
context.put(SystemConfig.TYPE, "Memory");
context.put(SystemConfig.CONTEXT, config);
return context;
}
-
- private File getFile(String name) {
- ClassLoader classLoader = getClass().getClassLoader();
- return new File(classLoader.getResource(name).getFile());
- }
}
diff --git a/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/sink/RabbitMQSinkTest.java b/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/sink/RabbitMQSinkTest.java
index 3b20c61f826..f03a36ce114 100644
--- a/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/sink/RabbitMQSinkTest.java
+++ b/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/sink/RabbitMQSinkTest.java
@@ -18,12 +18,15 @@
*/
package org.apache.pulsar.io.rabbitmq.sink;
+import static org.mockito.Mockito.mock;
import java.nio.charset.StandardCharsets;
+import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import org.apache.pulsar.functions.api.Record;
import org.apache.pulsar.functions.instance.SinkRecord;
+import org.apache.pulsar.io.core.SinkContext;
import org.apache.pulsar.io.rabbitmq.RabbitMQBrokerManager;
import org.apache.pulsar.io.rabbitmq.RabbitMQSink;
import org.awaitility.Awaitility;
@@ -46,7 +49,7 @@ public class RabbitMQSinkTest {
}
@Test
- public void TestOpenAndWriteSink() throws Exception {
+ public void testOpenAndWriteSink() throws Exception {
Map<String, Object> configs = new HashMap<>();
configs.put("host", "localhost");
configs.put("port", "5673");
@@ -66,7 +69,9 @@ public class RabbitMQSinkTest {
// open should success
// rabbitmq service may need time to initialize
- Awaitility.await().ignoreExceptions().untilAsserted(() -> sink.open(configs, null));
+ SinkContext sinkContext = mock(SinkContext.class);
+ Awaitility.await().ignoreExceptions().pollDelay(Duration.ofSeconds(1))
+ .untilAsserted(() -> sink.open(configs, sinkContext));
// write should success
Record<byte[]> record = build("test-topic", "fakeKey", "fakeValue", "fakeRoutingKey");
diff --git a/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/source/RabbitMQSourceTest.java b/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/source/RabbitMQSourceTest.java
index 2771185b841..08869e018c6 100644
--- a/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/source/RabbitMQSourceTest.java
+++ b/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/source/RabbitMQSourceTest.java
@@ -18,6 +18,9 @@
*/
package org.apache.pulsar.io.rabbitmq.source;
+import static org.mockito.Mockito.mock;
+import java.time.Duration;
+import org.apache.pulsar.io.core.SourceContext;
import org.apache.pulsar.io.rabbitmq.RabbitMQBrokerManager;
import org.apache.pulsar.io.rabbitmq.RabbitMQSource;
import org.awaitility.Awaitility;
@@ -44,7 +47,7 @@ public class RabbitMQSourceTest {
}
@Test
- public void TestOpenAndWriteSink() throws Exception {
+ public void testOpenAndWriteSink() throws Exception {
Map<String, Object> configs = new HashMap<>();
configs.put("host", "localhost");
configs.put("port", "5672");
@@ -66,8 +69,11 @@ public class RabbitMQSourceTest {
// open should success
// rabbitmq service may need time to initialize
- Awaitility.await().ignoreExceptions().untilAsserted(() -> source.open(configs, null));
+ SourceContext sourceContext = mock(SourceContext.class);
+ Awaitility.await().ignoreExceptions().pollDelay(Duration.ofSeconds(1))
+ .untilAsserted(() -> source.open(configs, sourceContext));
source.close();
}
+
}
diff --git a/pulsar-io/rabbitmq/src/test/resources/qpid.json b/pulsar-io/rabbitmq/src/test/resources/qpid.json
index 6a0381f6ddc..419e9cc1e4a 100644
--- a/pulsar-io/rabbitmq/src/test/resources/qpid.json
+++ b/pulsar-io/rabbitmq/src/test/resources/qpid.json
@@ -1,25 +1,57 @@
{
- "name": "EmbeddedBroker",
+ "name": "${broker.name}",
"modelVersion": "2.0",
- "storeVersion": 1,
"authenticationproviders": [
{
- "name": "noPassword",
- "type": "Anonymous",
- "secureOnlyMechanisms": []
- },
+ "name": "plain",
+ "type": "Plain",
+ "secureOnlyMechanisms": [],
+ "users": [
+ {
+ "name": "guest",
+ "password": "guest",
+ "type": "managed"
+ }
+ ]
+ }
+ ],
+ "brokerloggers": [
{
- "name": "passwordFile",
- "type": "PlainPasswordFile",
- "path": "${qpid.home_dir}${file.separator}etc${file.separator}passwd",
- "secureOnlyMechanisms": []
+ "name": "console",
+ "type": "Console",
+ "brokerloginclusionrules": [
+ {
+ "name": "Root",
+ "type": "NameAndLevel",
+ "level": "WARN",
+ "loggerName": "ROOT"
+ },
+ {
+ "name": "Qpid",
+ "type": "NameAndLevel",
+ "level": "INFO",
+ "loggerName": "org.apache.qpid.*"
+ },
+ {
+ "name": "Operational",
+ "type": "NameAndLevel",
+ "level": "INFO",
+ "loggerName": "qpid.message.*"
+ },
+ {
+ "name": "Statistics",
+ "type": "NameAndLevel",
+ "level": "INFO",
+ "loggerName": "qpid.statistics.*"
+ }
+ ]
}
],
"ports": [
{
"name": "AMQP",
"port": "${qpid.amqp_port}",
- "authenticationProvider": "passwordFile",
+ "authenticationProvider": "plain",
"protocols": [
"AMQP_0_9_1"
]
@@ -28,10 +60,9 @@
"virtualhostnodes": [
{
"name": "default",
- "type": "JSON",
+ "type": "Memory",
"defaultVirtualHostNode": "true",
- "virtualHostInitialConfiguration": "${qpid.initial_config_virtualhost_config}",
- "storeType": "DERBY"
+ "virtualHostInitialConfiguration": "{\"type\": \"Memory\"}"
}
]
}
\ No newline at end of file