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 2022/01/21 01:57:41 UTC
[pulsar] branch branch-2.9 updated (85bd12a -> 2ca5a04)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a change to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git.
from 85bd12a Fix SystemTopicBasedTopicPoliciesService NPE issue. (#13840)
new d3a2ee4 [Proxy] Prevent leak of unreleased lookupRequestSemaphore permits (#13812)
new 2dac261 [broker] Fix topic produced through REST not support Authorization (#13771)
new bf7af79 Getting rid of CVes brought in with aerospike (#13819)
new ff96162 Getting rid of CVEs in batch-data-generator (#13820)
new f68892c Getitng rid of CVEs in InfluxDB connector (#13821)
new 9909ec8 Getting rid of CVEs in Solr connector (#13822)
new 2ca5a04 [Broker] Avoid thread deadlock problem when creating topic policy reader (#13837)
The 7 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 | 6 +-
.../org/apache/pulsar/broker/rest/TopicsBase.java | 3 +-
.../SystemTopicBasedTopicPoliciesService.java | 12 +-
.../broker/service/TopicPoliciesService.java | 6 +-
.../apache/pulsar/broker/admin/TopicsAuthTest.java | 193 +++++++++++++++++++++
.../SystemTopicBasedTopicPoliciesServiceTest.java | 9 +-
.../systopic/PartitionedSystemTopicTest.java | 33 ++++
.../apache/pulsar/client/impl/RetryUtilTest.java | 13 +-
.../org/apache/pulsar/client/util/RetryUtil.java | 34 ++--
pulsar-io/batch-data-generator/pom.xml | 6 +
pulsar-io/influxdb/pom.xml | 4 +-
pulsar-io/solr/pom.xml | 2 +-
.../pulsar/proxy/server/LookupProxyHandler.java | 91 +++++-----
13 files changed, 316 insertions(+), 96 deletions(-)
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicsAuthTest.java
[pulsar] 01/07: [Proxy] Prevent leak of unreleased lookupRequestSemaphore permits (#13812)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d3a2ee444ecb54076b15dd41d69701658b435a4e
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Jan 19 06:41:12 2022 +0200
[Proxy] Prevent leak of unreleased lookupRequestSemaphore permits (#13812)
* [Proxy] Prevent leak of unreleased lookupRequestSemaphore permits
- should release permit in try-finally block
* Cleanup code in LookupProxyHandler
(cherry picked from commit 85b62e050b01b591a4b5751aab48b418ac9e4e76)
---
.../pulsar/proxy/server/LookupProxyHandler.java | 91 ++++++++++------------
1 file changed, 40 insertions(+), 51 deletions(-)
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java
index b14bea5..30e11a1 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java
@@ -18,14 +18,14 @@
*/
package org.apache.pulsar.proxy.server;
-import static org.apache.commons.lang3.StringUtils.isBlank;
-
+import io.netty.buffer.ByteBuf;
+import io.prometheus.client.Counter;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Optional;
-
+import java.util.concurrent.Semaphore;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.common.protocol.Commands;
@@ -42,13 +42,10 @@ import org.apache.pulsar.policies.data.loadbalancer.ServiceLookupData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import io.netty.buffer.ByteBuf;
-import io.prometheus.client.Counter;
-
public class LookupProxyHandler {
private final String throttlingErrorMessage = "Too many concurrent lookup and partitionsMetadata requests";
- private final ProxyService service;
private final ProxyConnection proxyConnection;
+ private final BrokerDiscoveryProvider discoveryProvider;
private final boolean connectWithTLS;
private SocketAddress clientAddress;
@@ -83,9 +80,11 @@ public class LookupProxyHandler {
.build("pulsar_proxy_rejected_get_topics_of_namespace_requests",
"Counter of getTopicsOfNamespace requests rejected due to throttling")
.create().register();
+ private final Semaphore lookupRequestSemaphore;
public LookupProxyHandler(ProxyService proxy, ProxyConnection proxyConnection) {
- this.service = proxy;
+ this.discoveryProvider = proxy.getDiscoveryProvider();
+ this.lookupRequestSemaphore = proxy.getLookupRequestSemaphore();
this.proxyConnection = proxyConnection;
this.clientAddress = proxyConnection.clientAddress();
this.connectWithTLS = proxy.getConfiguration().isTlsEnabledWithBroker();
@@ -98,28 +97,16 @@ public class LookupProxyHandler {
log.debug("Received Lookup from {}", clientAddress);
}
long clientRequestId = lookup.getRequestId();
- if (this.service.getLookupRequestSemaphore().tryAcquire()) {
- lookupRequests.inc();
- String topic = lookup.getTopic();
- String serviceUrl;
- if (isBlank(brokerServiceURL)) {
- ServiceLookupData availableBroker = null;
- try {
- availableBroker = service.getDiscoveryProvider().nextBroker();
- } catch (Exception e) {
- log.warn("[{}] Failed to get next active broker {}", clientAddress, e.getMessage(), e);
- proxyConnection.ctx().writeAndFlush(Commands.newLookupErrorResponse(ServerError.ServiceNotReady,
- e.getMessage(), clientRequestId));
- return;
+ if (lookupRequestSemaphore.tryAcquire()) {
+ try {
+ lookupRequests.inc();
+ String serviceUrl = getBrokerServiceUrl(clientRequestId);
+ if (serviceUrl != null) {
+ performLookup(clientRequestId, lookup.getTopic(), serviceUrl, false, 10);
}
- serviceUrl = this.connectWithTLS ? availableBroker.getPulsarServiceUrlTls()
- : availableBroker.getPulsarServiceUrl();
- } else {
- serviceUrl = this.connectWithTLS ? service.getConfiguration().getBrokerServiceURLTLS()
- : service.getConfiguration().getBrokerServiceURL();
+ } finally {
+ lookupRequestSemaphore.release();
}
- performLookup(clientRequestId, topic, serviceUrl, false, 10);
- this.service.getLookupRequestSemaphore().release();
} else {
rejectedLookupRequests.inc();
if (log.isDebugEnabled()) {
@@ -203,9 +190,12 @@ public class LookupProxyHandler {
log.debug("[{}] Received PartitionMetadataLookup", clientAddress);
}
final long clientRequestId = partitionMetadata.getRequestId();
- if (this.service.getLookupRequestSemaphore().tryAcquire()) {
- handlePartitionMetadataResponse(partitionMetadata, clientRequestId);
- this.service.getLookupRequestSemaphore().release();
+ if (lookupRequestSemaphore.tryAcquire()) {
+ try {
+ handlePartitionMetadataResponse(partitionMetadata, clientRequestId);
+ } finally {
+ lookupRequestSemaphore.release();
+ }
} else {
rejectedPartitionsMetadataRequests.inc();
if (log.isDebugEnabled()) {
@@ -277,9 +267,12 @@ public class LookupProxyHandler {
final long requestId = commandGetTopicsOfNamespace.getRequestId();
- if (this.service.getLookupRequestSemaphore().tryAcquire()) {
- handleGetTopicsOfNamespace(commandGetTopicsOfNamespace, requestId);
- this.service.getLookupRequestSemaphore().release();
+ if (lookupRequestSemaphore.tryAcquire()) {
+ try {
+ handleGetTopicsOfNamespace(commandGetTopicsOfNamespace, requestId);
+ } finally {
+ lookupRequestSemaphore.release();
+ }
} else {
rejectedGetTopicsOfNamespaceRequests.inc();
if (log.isDebugEnabled()) {
@@ -408,24 +401,20 @@ public class LookupProxyHandler {
* Get default broker service url or discovery an available broker
**/
private String getBrokerServiceUrl(long clientRequestId) {
- if (isBlank(brokerServiceURL)) {
- ServiceLookupData availableBroker;
- try {
- availableBroker = service.getDiscoveryProvider().nextBroker();
- } catch (Exception e) {
- log.warn("[{}] Failed to get next active broker {}", clientAddress, e.getMessage(), e);
- proxyConnection.ctx().writeAndFlush(Commands.newError(
- clientRequestId, ServerError.ServiceNotReady, e.getMessage()
- ));
- return null;
- }
- return this.connectWithTLS ?
- availableBroker.getPulsarServiceUrlTls() : availableBroker.getPulsarServiceUrl();
- } else {
- return this.connectWithTLS ?
- service.getConfiguration().getBrokerServiceURLTLS() : service.getConfiguration().getBrokerServiceURL();
+ if (StringUtils.isNotBlank(brokerServiceURL)) {
+ return brokerServiceURL;
}
-
+ ServiceLookupData availableBroker;
+ try {
+ availableBroker = discoveryProvider.nextBroker();
+ } catch (Exception e) {
+ log.warn("[{}] Failed to get next active broker {}", clientAddress, e.getMessage(), e);
+ proxyConnection.ctx().writeAndFlush(Commands.newError(
+ clientRequestId, ServerError.ServiceNotReady, e.getMessage()
+ ));
+ return null;
+ }
+ return this.connectWithTLS ? availableBroker.getPulsarServiceUrlTls() : availableBroker.getPulsarServiceUrl();
}
private InetSocketAddress getAddr(String brokerServiceUrl, long clientRequestId) {
[pulsar] 03/07: Getting rid of CVes brought in with aerospike (#13819)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit bf7af7943895377a465943a19c93ba191920edea
Author: Andrey Yegorov <86...@users.noreply.github.com>
AuthorDate: Wed Jan 19 23:51:31 2022 -0800
Getting rid of CVes brought in with aerospike (#13819)
CVE-2020-26939
CVE-2015-0886
(cherry picked from commit 8a452b733c91d6a60ddf32b403c9169be5ba7978)
---
pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pom.xml b/pom.xml
index 91a9a4a..cfa1c2d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -138,7 +138,7 @@ flexible messaging model and an intuitive client API.</description>
<sketches.version>0.8.3</sketches.version>
<hbc-core.version>2.2.0</hbc-core.version>
<cassandra-driver-core.version>3.6.0</cassandra-driver-core.version>
- <aerospike-client.version>4.4.8</aerospike-client.version>
+ <aerospike-client.version>4.4.20</aerospike-client.version>
<kafka-client.version>2.7.0</kafka-client.version>
<rabbitmq-client.version>5.1.1</rabbitmq-client.version>
<aws-sdk.version>1.11.774</aws-sdk.version>
[pulsar] 07/07: [Broker] Avoid thread deadlock problem when creating topic policy reader (#13837)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 2ca5a0469a466510ff801a8e737430ee47edaf39
Author: ran <ga...@126.com>
AuthorDate: Thu Jan 20 20:50:53 2022 +0800
[Broker] Avoid thread deadlock problem when creating topic policy reader (#13837)
Currently, the topic policy reader creation thread is the `executor` of the `PulsarService`, this thread is also used to search the candidate broker. If they use the same thread in some conditions, the lookup request will be blocked and result in a lookup request timeout.
We could find out that the lookup request was blocked 1 minute until the lookup request timeout. The thread `pulsar-2-8` was blocked by topic policy reader creation.
Change the topic policy reader creation to be asynchronous. Modify the method `RetryUtil.retryAsynchronously` to handle asynchronous execution.
Add a new test to verify consumer creations can be successful when enabling the topic policy feature.
(cherry picked from commit 760bfec51ed6f043024c3c42bc73639fc2c117bb)
---
.../SystemTopicBasedTopicPoliciesService.java | 12 ++------
.../broker/service/TopicPoliciesService.java | 6 ++--
.../SystemTopicBasedTopicPoliciesServiceTest.java | 9 +++---
.../systopic/PartitionedSystemTopicTest.java | 33 +++++++++++++++++++++
.../apache/pulsar/client/impl/RetryUtilTest.java | 13 +++++----
.../org/apache/pulsar/client/util/RetryUtil.java | 34 ++++++++++------------
6 files changed, 69 insertions(+), 38 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
index 44b866f..838edfe 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
@@ -231,7 +231,7 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
private void prepareInitPoliciesCache(NamespaceName namespace, CompletableFuture<Void> result) {
if (policyCacheInitMap.putIfAbsent(namespace, false) == null) {
CompletableFuture<SystemTopicClient.Reader<PulsarEvent>> readerCompletableFuture =
- creatSystemTopicClientWithRetry(namespace);
+ createSystemTopicClientWithRetry(namespace);
readerCaches.put(namespace, readerCompletableFuture);
ownedBundlesCountPerNamespace.putIfAbsent(namespace, new AtomicInteger(1));
readerCompletableFuture.whenComplete((reader, ex) -> {
@@ -248,7 +248,7 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
}
}
- protected CompletableFuture<SystemTopicClient.Reader<PulsarEvent>> creatSystemTopicClientWithRetry(
+ protected CompletableFuture<SystemTopicClient.Reader<PulsarEvent>> createSystemTopicClientWithRetry(
NamespaceName namespace) {
CompletableFuture<SystemTopicClient.Reader<PulsarEvent>> result = new CompletableFuture<>();
try {
@@ -260,13 +260,7 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
SystemTopicClient<PulsarEvent> systemTopicClient = namespaceEventsSystemTopicFactory
.createTopicPoliciesSystemTopicClient(namespace);
Backoff backoff = new Backoff(1, TimeUnit.SECONDS, 3, TimeUnit.SECONDS, 10, TimeUnit.SECONDS);
- RetryUtil.retryAsynchronously(() -> {
- try {
- return systemTopicClient.newReader();
- } catch (PulsarClientException e) {
- throw new RuntimeException(e);
- }
- }, backoff, pulsarService.getExecutor(), result);
+ RetryUtil.retryAsynchronously(systemTopicClient::newReaderAsync, backoff, pulsarService.getExecutor(), result);
return result;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPoliciesService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPoliciesService.java
index 53de087..0165dc1 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPoliciesService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPoliciesService.java
@@ -85,11 +85,13 @@ public interface TopicPoliciesService {
.create() : backoff;
try {
RetryUtil.retryAsynchronously(() -> {
+ CompletableFuture<Optional<TopicPolicies>> future = new CompletableFuture<>();
try {
- return Optional.ofNullable(getTopicPolicies(topicName));
+ future.complete(Optional.ofNullable(getTopicPolicies(topicName)));
} catch (BrokerServiceException.TopicPoliciesCacheNotInitException exception) {
- throw new RuntimeException(exception);
+ future.completeExceptionally(exception);
}
+ return future;
}, usedBackoff, scheduledExecutorService, response);
} catch (Exception e) {
response.completeExceptionally(e);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java
index 9a489cf..8b5f420 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java
@@ -20,7 +20,6 @@ package org.apache.pulsar.broker.service;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;
@@ -56,6 +55,7 @@ import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.policies.data.TopicPolicies;
+import org.apache.pulsar.common.util.FutureUtil;
import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
@@ -313,7 +313,7 @@ public class SystemTopicBasedTopicPoliciesServiceTest extends MockedPulsarServic
try {
service.getTopicPoliciesAsyncWithRetry(TOPIC1, backoff, pulsar.getExecutor()).get();
} catch (Exception e) {
- assertTrue(e.getCause().getCause() instanceof TopicPoliciesCacheNotInitException);
+ assertTrue(e.getCause() instanceof TopicPoliciesCacheNotInitException);
}
long cost = System.currentTimeMillis() - start;
assertTrue("actual:" + cost, cost >= 5000 - 1000);
@@ -333,9 +333,10 @@ public class SystemTopicBasedTopicPoliciesServiceTest extends MockedPulsarServic
SystemTopicClient.Reader<PulsarEvent> reader = mock(SystemTopicClient.Reader.class);
// Throw an exception first, create successfully after retrying
- doThrow(new PulsarClientException("test")).doReturn(reader).when(client).newReader();
+ doReturn(FutureUtil.failedFuture(new PulsarClientException("test")))
+ .doReturn(CompletableFuture.completedFuture(reader)).when(client).newReaderAsync();
- SystemTopicClient.Reader<PulsarEvent> reader1 = service.creatSystemTopicClientWithRetry(null).get();
+ SystemTopicClient.Reader<PulsarEvent> reader1 = service.createSystemTopicClientWithRetry(null).get();
assertEquals(reader1, reader);
}
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 bd4ef78..7dd02bd 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
@@ -18,14 +18,23 @@
*/
package org.apache.pulsar.broker.systopic;
+import com.google.common.collect.Sets;
+import org.apache.commons.lang.RandomStringUtils;
import org.apache.pulsar.broker.service.BrokerTestBase;
+import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.common.events.EventsTopicNames;
import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.common.util.FutureUtil;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
@Test(groups = "broker")
public class PartitionedSystemTopicTest extends BrokerTestBase {
@@ -66,4 +75,28 @@ public class PartitionedSystemTopicTest extends BrokerTestBase {
Assert.assertEquals(partitions, PARTITIONS);
Assert.assertEquals(admin.topics().getList(ns).size(), PARTITIONS);
}
+
+ @Test(timeOut = 1000 * 60)
+ public void testConsumerCreationWhenEnablingTopicPolicy() throws Exception {
+ String tenant = "tenant-" + RandomStringUtils.randomAlphabetic(4).toLowerCase();
+ admin.tenants().createTenant(tenant, new TenantInfoImpl(Sets.newHashSet(), Sets.newHashSet("test")));
+ int namespaceCount = 30;
+ for (int i = 0; i < namespaceCount; i++) {
+ String ns = tenant + "/ns-" + i;
+ admin.namespaces().createNamespace(ns, 4);
+ String topic = ns + "/t1";
+ admin.topics().createPartitionedTopic(topic, 2);
+ }
+
+ List<CompletableFuture<Consumer<byte[]>>> futureList = new ArrayList<>();
+ for (int i = 0; i < namespaceCount; i++) {
+ String topic = tenant + "/ns-" + i + "/t1";
+ futureList.add(pulsarClient.newConsumer()
+ .topic(topic)
+ .subscriptionName("sub")
+ .subscribeAsync());
+ }
+ FutureUtil.waitForAll(futureList).get();
+ }
+
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RetryUtilTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RetryUtilTest.java
index e17f376..f554430 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RetryUtilTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RetryUtilTest.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.client.impl;
import org.apache.pulsar.client.util.RetryUtil;
+import org.apache.pulsar.common.util.FutureUtil;
import org.testng.annotations.Test;
import java.util.concurrent.CompletableFuture;
@@ -45,11 +46,14 @@ public class RetryUtilTest {
.setMandatoryStop(5000, TimeUnit.MILLISECONDS)
.create();
RetryUtil.retryAsynchronously(() -> {
+ CompletableFuture<Boolean> future = new CompletableFuture<>();
atomicInteger.incrementAndGet();
if (atomicInteger.get() < 5) {
- throw new RuntimeException("fail");
+ future.completeExceptionally(new RuntimeException("fail"));
+ } else {
+ future.complete(true);
}
- return true;
+ return future;
}, backoff, executor, callback);
assertTrue(callback.get());
assertEquals(atomicInteger.get(), 5);
@@ -66,9 +70,8 @@ public class RetryUtilTest {
.setMandatoryStop(5000, TimeUnit.MILLISECONDS)
.create();
long start = System.currentTimeMillis();
- RetryUtil.retryAsynchronously(() -> {
- throw new RuntimeException("fail");
- }, backoff, executor, callback);
+ RetryUtil.retryAsynchronously(() ->
+ FutureUtil.failedFuture(new RuntimeException("fail")), backoff, executor, callback);
try {
callback.get();
} catch (Exception e) {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/util/RetryUtil.java b/pulsar-client/src/main/java/org/apache/pulsar/client/util/RetryUtil.java
index 084a583..b3ed2c3 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/util/RetryUtil.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/util/RetryUtil.java
@@ -30,7 +30,7 @@ import java.util.function.Supplier;
public class RetryUtil {
private static final Logger log = LoggerFactory.getLogger(RetryUtil.class);
- public static <T> void retryAsynchronously(Supplier<T> supplier, Backoff backoff,
+ public static <T> void retryAsynchronously(Supplier<CompletableFuture<T>> supplier, Backoff backoff,
ScheduledExecutorService scheduledExecutorService,
CompletableFuture<T> callback) {
if (backoff.getMax() <= 0) {
@@ -43,26 +43,24 @@ public class RetryUtil {
executeWithRetry(supplier, backoff, scheduledExecutorService, callback));
}
- private static <T> void executeWithRetry(Supplier<T> supplier, Backoff backoff,
+ private static <T> void executeWithRetry(Supplier<CompletableFuture<T>> supplier, Backoff backoff,
ScheduledExecutorService scheduledExecutorService,
CompletableFuture<T> callback) {
- try {
- T result = supplier.get();
- callback.complete(result);
- } catch (Exception e) {
- long next = backoff.next();
- boolean isMandatoryStop = backoff.isMandatoryStopMade();
- if (isMandatoryStop) {
- callback.completeExceptionally(e);
- } else {
- if (log.isDebugEnabled()) {
- log.debug("execute with retry fail, will retry in {} ms", next, e);
+ supplier.get().whenComplete((result, e) -> {
+ if (e != null) {
+ long next = backoff.next();
+ boolean isMandatoryStop = backoff.isMandatoryStopMade();
+ if (isMandatoryStop) {
+ callback.completeExceptionally(e);
+ } else {
+ log.warn("Execution with retry fail, because of {}, will retry in {} ms", e.getMessage(), next);
+ scheduledExecutorService.schedule(() ->
+ executeWithRetry(supplier, backoff, scheduledExecutorService, callback),
+ next, TimeUnit.MILLISECONDS);
}
- log.info("Because of {} , will retry in {} ms", e.getMessage(), next);
- scheduledExecutorService.schedule(() ->
- executeWithRetry(supplier, backoff, scheduledExecutorService, callback),
- next, TimeUnit.MILLISECONDS);
+ return;
}
- }
+ callback.complete(result);
+ });
}
}
[pulsar] 02/07: [broker] Fix topic produced through REST not support Authorization (#13771)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 2dac261ea55d604a39cc82593818a13639725c5c
Author: liudezhi <33...@users.noreply.github.com>
AuthorDate: Thu Jan 20 15:35:23 2022 +0800
[broker] Fix topic produced through REST not support Authorization (#13771)
(cherry picked from commit f551fdc1d34570bc28ae1eccb9c409f02a6e00dd)
---
.../org/apache/pulsar/broker/rest/TopicsBase.java | 3 +-
.../apache/pulsar/broker/admin/TopicsAuthTest.java | 193 +++++++++++++++++++++
2 files changed, 195 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/rest/TopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/rest/TopicsBase.java
index 1df2c14..f89abf9 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/rest/TopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/rest/TopicsBase.java
@@ -756,7 +756,8 @@ public class TopicsBase extends PersistentTopicsBase {
}
boolean isAuthorized = pulsar().getBrokerService().getAuthorizationService()
- .canProduce(topicName, originalPrincipal(), clientAuthData());
+ .canProduce(topicName, originalPrincipal() == null ? clientAppId() : originalPrincipal(),
+ clientAuthData());
if (!isAuthorized) {
throw new RestException(Status.UNAUTHORIZED, String.format("Unauthorized to produce to topic %s"
+ " with clientAppId [%s] and authdata %s", topicName.toString(),
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicsAuthTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicsAuthTest.java
new file mode 100644
index 0000000..185053e
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicsAuthTest.java
@@ -0,0 +1,193 @@
+/**
+ * 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.admin;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.google.common.collect.Sets;
+import io.jsonwebtoken.Jwts;
+import io.jsonwebtoken.SignatureAlgorithm;
+import java.util.Base64;
+import java.util.EnumSet;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import javax.crypto.SecretKey;
+import javax.ws.rs.client.Client;
+import javax.ws.rs.client.ClientBuilder;
+import javax.ws.rs.client.Entity;
+import javax.ws.rs.client.WebTarget;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.broker.authentication.AuthenticationProviderToken;
+import org.apache.pulsar.broker.authentication.utils.AuthTokenUtils;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminBuilder;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.auth.AuthenticationToken;
+import org.apache.pulsar.client.impl.schema.StringSchema;
+import org.apache.pulsar.common.policies.data.AuthAction;
+import org.apache.pulsar.common.policies.data.ClusterDataImpl;
+import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.common.util.ObjectMapperFactory;
+import org.apache.pulsar.websocket.data.ProducerMessage;
+import org.apache.pulsar.websocket.data.ProducerMessages;
+import org.glassfish.jersey.client.ClientConfig;
+import org.glassfish.jersey.client.ClientProperties;
+import org.glassfish.jersey.media.multipart.MultiPartFeature;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+public class TopicsAuthTest extends MockedPulsarServiceBaseTest {
+
+ private final String testLocalCluster = "test";
+ private final String testTenant = "my-tenant";
+ private final String testNamespace = "my-namespace";
+ private final String testTopicName = "my-topic";
+
+ private static final SecretKey SECRET_KEY = AuthTokenUtils.createSecretKey(SignatureAlgorithm.HS256);
+ private static final String ADMIN_TOKEN = Jwts.builder().setSubject("admin").signWith(SECRET_KEY).compact();
+ private static final String PRODUCE_TOKEN = Jwts.builder().setSubject("producer").signWith(SECRET_KEY).compact();
+ private static final String CONSUME_TOKEN = Jwts.builder().setSubject("consumer").signWith(SECRET_KEY).compact();
+
+ @Override
+ @BeforeMethod
+ protected void setup() throws Exception {
+ // enable auth&auth and use JWT at broker
+ conf.setAuthenticationEnabled(true);
+ conf.setAuthorizationEnabled(true);
+ conf.getProperties().setProperty("tokenSecretKey", "data:;base64,"
+ + Base64.getEncoder().encodeToString(SECRET_KEY.getEncoded()));
+ Set<String> superUserRoles = new HashSet<>();
+ superUserRoles.add("admin");
+ conf.setSuperUserRoles(superUserRoles);
+ Set<String> providers = new HashSet<>();
+ providers.add(AuthenticationProviderToken.class.getName());
+ conf.setAuthenticationProviders(providers);
+ super.internalSetup();
+ PulsarAdminBuilder pulsarAdminBuilder = PulsarAdmin.builder().serviceHttpUrl(brokerUrl != null
+ ? brokerUrl.toString() : brokerUrlTls.toString())
+ .authentication(AuthenticationToken.class.getName(),
+ ADMIN_TOKEN);
+ admin = Mockito.spy(pulsarAdminBuilder.build());
+ admin.clusters().createCluster(testLocalCluster, new ClusterDataImpl());
+ admin.tenants().createTenant(testTenant, new TenantInfoImpl(Sets.newHashSet("role1", "role2"),
+ Sets.newHashSet(testLocalCluster)));
+ admin.namespaces().createNamespace(testTenant + "/" + testNamespace,
+ Sets.newHashSet(testLocalCluster));
+ admin.namespaces().grantPermissionOnNamespace(testTenant + "/" + testNamespace, "producer",
+ EnumSet.of(AuthAction.produce));
+ admin.namespaces().grantPermissionOnNamespace(testTenant + "/" + testNamespace, "consumer",
+ EnumSet.of(AuthAction.consume));
+ }
+
+ @Override
+ @AfterMethod
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @DataProvider(name = "variations")
+ public static Object[][] variations() {
+ return new Object[][]{
+ {CONSUME_TOKEN, 401},
+ {PRODUCE_TOKEN, 200}
+ };
+ }
+
+ @Test(dataProvider = "variations")
+ public void testProduceToNonPartitionedTopic(String token, int status) throws Exception {
+ innerTestProduce(testTopicName, true, false, token, status);
+ }
+
+ @Test(dataProvider = "variations")
+ public void testProduceToPartitionedTopic(String token, int status) throws Exception {
+ innerTestProduce(testTopicName, true, true, token, status);
+ }
+
+ @Test(dataProvider = "variations")
+ public void testProduceOnNonPersistentNonPartitionedTopic(String token, int status) throws Exception {
+ innerTestProduce(testTopicName, false, false, token, status);
+ }
+
+ @Test(dataProvider = "variations")
+ public void testProduceOnNonPersistentPartitionedTopic(String token, int status) throws Exception {
+ innerTestProduce(testTopicName, false, true, token, status);
+ }
+
+ private void innerTestProduce(String createTopicName, boolean isPersistent, boolean isPartition,
+ String token, int status) throws Exception {
+ String topicPrefix = null;
+ if (isPersistent == true) {
+ topicPrefix = "persistent";
+ } else {
+ topicPrefix = "non-persistent";
+ }
+ if (isPartition == true) {
+ admin.topics().createPartitionedTopic(topicPrefix + "://" + testTenant + "/"
+ + testNamespace + "/" + createTopicName, 5);
+ } else {
+ admin.topics().createNonPartitionedTopic(topicPrefix + "://" + testTenant + "/"
+ + testNamespace + "/" + createTopicName);
+ }
+ Schema<String> schema = StringSchema.utf8();
+ ProducerMessages producerMessages = new ProducerMessages();
+ producerMessages.setKeySchema(ObjectMapperFactory.getThreadLocal().
+ writeValueAsString(schema.getSchemaInfo()));
+ producerMessages.setValueSchema(ObjectMapperFactory.getThreadLocal().
+ writeValueAsString(schema.getSchemaInfo()));
+ String message = "[" +
+ "{\"key\":\"my-key\",\"payload\":\"RestProducer:1\",\"eventTime\":1603045262772,\"sequenceId\":1}," +
+ "{\"key\":\"my-key\",\"payload\":\"RestProducer:2\",\"eventTime\":1603045262772,\"sequenceId\":2}]";
+ producerMessages.setMessages(ObjectMapperFactory.getThreadLocal().readValue(message,
+ new TypeReference<List<ProducerMessage>>() {
+ }));
+
+ WebTarget root = buildWebClient();
+ String requestPath = null;
+ if (isPartition == true) {
+ requestPath = "/topics/" + topicPrefix + "/" + testTenant + "/" + testNamespace + "/"
+ + createTopicName + "/partitions/2";
+ } else {
+ requestPath = "/topics/" + topicPrefix + "/" + testTenant + "/" + testNamespace + "/" + createTopicName;
+ }
+
+ Response response = root.path(requestPath)
+ .request(MediaType.APPLICATION_JSON)
+ .header("Authorization", "Bearer " + token)
+ .post(Entity.json(producerMessages));
+ Assert.assertEquals(response.getStatus(), status);
+ }
+
+ WebTarget buildWebClient() throws Exception {
+ ClientConfig httpConfig = new ClientConfig();
+ httpConfig.property(ClientProperties.FOLLOW_REDIRECTS, true);
+ httpConfig.property(ClientProperties.ASYNC_THREADPOOL_SIZE, 8);
+ httpConfig.register(MultiPartFeature.class);
+
+ javax.ws.rs.client.ClientBuilder clientBuilder = ClientBuilder.newBuilder().withConfig(httpConfig);
+ Client client = clientBuilder.build();
+ return client.target(brokerUrl.toString());
+ }
+
+}
[pulsar] 04/07: Getting rid of CVEs in batch-data-generator (#13820)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ff96162732b6978d8046ddb6caa28d62f5bf101a
Author: Andrey Yegorov <86...@users.noreply.github.com>
AuthorDate: Wed Jan 19 23:51:57 2022 -0800
Getting rid of CVEs in batch-data-generator (#13820)
CVE-2021-41269
CVE-2021-22060
CVE-2021-22096
CVE-2021-22118
(cherry picked from commit 5a33d106dd282fa85008888a687c94203f888df7)
---
pom.xml | 4 ++--
pulsar-io/batch-data-generator/pom.xml | 6 ++++++
2 files changed, 8 insertions(+), 2 deletions(-)
diff --git a/pom.xml b/pom.xml
index cfa1c2d..22afc5c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -198,8 +198,8 @@ flexible messaging model and an intuitive client API.</description>
<!-- override kotlin-stdlib used by okio in order to address CVE-2020-29582 -->
<kotlin-stdlib.version>1.4.32</kotlin-stdlib.version>
<nsq-client.version>1.0</nsq-client.version>
- <cron-utils.version>9.1.3</cron-utils.version>
- <spring-context.version>5.3.1</spring-context.version>
+ <cron-utils.version>9.1.6</cron-utils.version>
+ <spring-context.version>5.3.15</spring-context.version>
<apache-http-client.version>4.5.13</apache-http-client.version>
<!-- test dependencies -->
diff --git a/pulsar-io/batch-data-generator/pom.xml b/pulsar-io/batch-data-generator/pom.xml
index 8141194..288583b 100644
--- a/pulsar-io/batch-data-generator/pom.xml
+++ b/pulsar-io/batch-data-generator/pom.xml
@@ -45,6 +45,12 @@
</dependency>
<dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-context</artifactId>
+ <version>${spring-context.version}</version>
+ </dependency>
+
+ <dependency>
<groupId>io.codearte.jfairy</groupId>
<artifactId>jfairy</artifactId>
<version>0.5.9</version>
[pulsar] 05/07: Getitng rid of CVEs in InfluxDB connector (#13821)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit f68892c03c4393f7105878a6ca96268a4a206b7d
Author: Andrey Yegorov <86...@users.noreply.github.com>
AuthorDate: Wed Jan 19 23:52:23 2022 -0800
Getitng rid of CVEs in InfluxDB connector (#13821)
CVE-2019-20933
CVE-2018-1000850
(cherry picked from commit 85746390c4421cce710ec6988d1747fd5a3f7ade)
---
pulsar-io/influxdb/pom.xml | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/pulsar-io/influxdb/pom.xml b/pulsar-io/influxdb/pom.xml
index 8c7c6d0..f95f19a 100644
--- a/pulsar-io/influxdb/pom.xml
+++ b/pulsar-io/influxdb/pom.xml
@@ -51,13 +51,13 @@
<dependency>
<groupId>com.influxdb</groupId>
<artifactId>influxdb-client-java</artifactId>
- <version>1.6.0</version>
+ <version>4.0.0</version>
</dependency>
<dependency>
<groupId>org.influxdb</groupId>
<artifactId>influxdb-java</artifactId>
- <version>2.7</version>
+ <version>2.22</version>
<exclusions>
<exclusion>
<groupId>com.squareup.okhttp3</groupId>
[pulsar] 06/07: Getting rid of CVEs in Solr connector (#13822)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9909ec859a5668005591a105565b4abde87b3341
Author: Andrey Yegorov <86...@users.noreply.github.com>
AuthorDate: Wed Jan 19 23:52:46 2022 -0800
Getting rid of CVEs in Solr connector (#13822)
CVE-2019-17638
CVE-2020-27216
CVE-2020-27218
CVE-2020-27223
CVE-2021-28165
CVE-2021-28169
CVE-2021-34428
CVE-2021-27905
CVE-2021-29262
CVE-2021-29943
CVE-2021-44548
(cherry picked from commit 76fc2fbc9801da5232cd4ba32d25298b74d460db)
---
pulsar-io/solr/pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pulsar-io/solr/pom.xml b/pulsar-io/solr/pom.xml
index db3cafc..615d476 100644
--- a/pulsar-io/solr/pom.xml
+++ b/pulsar-io/solr/pom.xml
@@ -29,7 +29,7 @@
</parent>
<properties>
- <solr.version>8.6.3</solr.version>
+ <solr.version>8.11.1</solr.version>
</properties>
<artifactId>pulsar-io-solr</artifactId>