From 209e4241c2df251c33fdbbbd5de0c22e50572f33 Mon Sep 17 00:00:00 2001 From: mattison chao Date: Tue, 15 Feb 2022 13:42:03 +0800 Subject: [PATCH 1/6] [Branch 2.7] Fix get topic policy timeout. --- .../pulsar/broker/admin/AdminResource.java | 59 ++++++------- .../SystemTopicBasedTopicPoliciesService.java | 54 ++++++++---- .../apache/pulsar/client/util/RetryUtil.java | 67 +++++++++++++++ .../pulsar/client/util/RetryUtilTest.java | 85 +++++++++++++++++++ 4 files changed, 216 insertions(+), 49 deletions(-) create mode 100644 pulsar-client/src/main/java/org/apache/pulsar/client/util/RetryUtil.java create mode 100644 pulsar-client/src/test/java/org/apache/pulsar/client/util/RetryUtilTest.java diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java index f0c028a8146ce..6fd4419bbac10 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java @@ -33,10 +33,9 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Collectors; import javax.servlet.ServletContext; import javax.ws.rs.WebApplicationException; @@ -44,7 +43,6 @@ import javax.ws.rs.core.Response; import javax.ws.rs.core.Response.Status; import javax.ws.rs.core.UriBuilder; - import com.google.errorprone.annotations.CanIgnoreReturnValue; import org.apache.bookkeeper.util.ZkUtils; import org.apache.pulsar.broker.PulsarService; @@ -56,6 +54,7 @@ import org.apache.pulsar.client.admin.internal.TopicsImpl; import org.apache.pulsar.client.impl.Backoff; import org.apache.pulsar.client.impl.BackoffBuilder; +import org.apache.pulsar.client.util.RetryUtil; import org.apache.pulsar.common.api.proto.PulsarApi; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.common.naming.Constants; @@ -552,44 +551,36 @@ protected BacklogQuota namespaceBacklogQuota(String namespace, String namespaceP } protected CompletableFuture> getTopicPoliciesAsyncWithRetry(TopicName topicName) { - return internalGetTopicPoliciesAsyncWithRetry(topicName, - new AtomicLong(DEFAULT_GET_TOPIC_POLICY_TIMEOUT), null, null); + try { + checkTopicLevelPolicyEnable(); + }catch (RestException ex) { + log.error("[{}] Failed to get topic policies {}", clientAppId(), topicName, ex); + return FutureUtil.failedFuture(ex); + } + return internalGetTopicPoliciesAsyncWithRetry(topicName, null, pulsar().getExecutor()); } protected CompletableFuture> internalGetTopicPoliciesAsyncWithRetry(TopicName topicName, - final AtomicLong remainingTime, final Backoff backoff, - CompletableFuture> future) { - CompletableFuture> response = future == null ? new CompletableFuture<>() : future; + ScheduledExecutorService scheduledExecutorService) { + CompletableFuture> response = new CompletableFuture<>(); + Backoff usedBackoff = backoff == null ? new BackoffBuilder() + .setInitialTime(500, TimeUnit.MILLISECONDS) + .setMandatoryStop(DEFAULT_GET_TOPIC_POLICY_TIMEOUT, TimeUnit.MILLISECONDS) + .setMax(DEFAULT_GET_TOPIC_POLICY_TIMEOUT, TimeUnit.MILLISECONDS) + .create() : backoff; try { - checkTopicLevelPolicyEnable(); - response.complete(Optional.ofNullable(pulsar() - .getTopicPoliciesService().getTopicPolicies(topicName))); - } catch (RestException re) { - response.completeExceptionally(re); - } catch (BrokerServiceException.TopicPoliciesCacheNotInitException e) { - Backoff usedBackoff = backoff == null ? new BackoffBuilder() - .setInitialTime(500, TimeUnit.MILLISECONDS) - .setMandatoryStop(DEFAULT_GET_TOPIC_POLICY_TIMEOUT, TimeUnit.MILLISECONDS) - .setMax(DEFAULT_GET_TOPIC_POLICY_TIMEOUT, TimeUnit.MILLISECONDS) - .create() : backoff; - long nextDelay = Math.min(usedBackoff.next(), remainingTime.get()); - if (nextDelay <= 0) { - response.completeExceptionally(new TimeoutException( - String.format("Failed to get topic policy withing configured timeout %s ms", - DEFAULT_GET_TOPIC_POLICY_TIMEOUT))); - } else { - if (log.isDebugEnabled()) { - log.error("Topic {} policies have not been initialized yet, retry after {}ms", - topicName, nextDelay); + RetryUtil.retryAsynchronously(() -> { + CompletableFuture> future = new CompletableFuture<>(); + try { + future.complete(Optional.ofNullable(pulsar().getTopicPoliciesService() + .getTopicPolicies(topicName))); + } catch (BrokerServiceException.TopicPoliciesCacheNotInitException exception) { + future.completeExceptionally(exception); } - pulsar().getExecutor().schedule(() -> { - remainingTime.addAndGet(-nextDelay); - internalGetTopicPoliciesAsyncWithRetry(topicName, remainingTime, usedBackoff, response); - }, nextDelay, TimeUnit.MILLISECONDS); - } + return future; + }, usedBackoff, scheduledExecutorService, response); } catch (Exception e) { - log.error("[{}] Failed to get topic policies {}", clientAppId(), topicName, e); response.completeExceptionally(e); } return response; 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 3dc5a1014cb67..4448b72dca654 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 @@ -25,6 +25,7 @@ import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.apache.pulsar.broker.PulsarServerException; import org.apache.pulsar.broker.PulsarService; @@ -34,6 +35,8 @@ import org.apache.pulsar.broker.systopic.SystemTopicClient; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.impl.Backoff; +import org.apache.pulsar.client.util.RetryUtil; import org.apache.pulsar.common.events.ActionType; import org.apache.pulsar.common.events.EventType; import org.apache.pulsar.common.events.PulsarEvent; @@ -144,6 +147,10 @@ public boolean cacheIsInitialized(TopicName topicName) { @Override public TopicPolicies getTopicPolicies(TopicName topicName) throws TopicPoliciesCacheNotInitException { + if (!policyCacheInitMap.containsKey(topicName.getNamespaceObject())) { + NamespaceName namespace = topicName.getNamespaceObject(); + prepareInitPoliciesCache(namespace, new CompletableFuture<>()); + } if (policyCacheInitMap.containsKey(topicName.getNamespaceObject()) && !policyCacheInitMap.get(topicName.getNamespaceObject())) { throw new TopicPoliciesCacheNotInitException(); @@ -151,6 +158,25 @@ public TopicPolicies getTopicPolicies(TopicName topicName) throws TopicPoliciesC return policiesCache.get(TopicName.get(topicName.getPartitionedTopicName())); } + private void prepareInitPoliciesCache(NamespaceName namespace, CompletableFuture result) { + if (policyCacheInitMap.putIfAbsent(namespace, false) == null) { + CompletableFuture readerCompletableFuture = + createSystemTopicClientWithRetry(namespace); + readerCaches.put(namespace, readerCompletableFuture); + readerCompletableFuture.whenComplete((reader, ex) -> { + if (ex != null) { + log.error("[{}] Failed to create reader on __change_events topic", namespace, ex); + result.completeExceptionally(ex); + readerCaches.remove(namespace); + reader.closeAsync(); + } else { + initPolicesCache(reader, result); + result.thenRun(() -> readMorePolicies(reader)); + } + }); + } + } + @Override public CompletableFuture getTopicPoliciesBypassCacheAsync(TopicName topicName) { CompletableFuture result = new CompletableFuture<>(); @@ -174,32 +200,30 @@ public CompletableFuture addOwnedNamespaceBundleAsync(NamespaceBundle name result.complete(null); return result; } - createSystemTopicFactoryIfNeeded(); synchronized (this) { if (readerCaches.get(namespace) != null) { ownedBundlesCountPerNamespace.get(namespace).incrementAndGet(); result.complete(null); + return result; } else { - SystemTopicClient systemTopicClient = namespaceEventsSystemTopicFactory.createSystemTopic(namespace - , EventType.TOPIC_POLICY); ownedBundlesCountPerNamespace.putIfAbsent(namespace, new AtomicInteger(1)); - policyCacheInitMap.put(namespace, false); - CompletableFuture readerCompletableFuture = systemTopicClient.newReaderAsync(); - readerCaches.put(namespace, readerCompletableFuture); - readerCompletableFuture.whenComplete((reader, ex) -> { - if (ex != null) { - log.error("[{}] Failed to create reader on __change_events topic", namespace, ex); - result.completeExceptionally(ex); - } else { - initPolicesCache(reader, result); - result.thenRun(() -> readMorePolicies(reader)); - } - }); + prepareInitPoliciesCache(namespace, result); } } return result; } + protected CompletableFuture createSystemTopicClientWithRetry( + NamespaceName namespace) { + CompletableFuture result = new CompletableFuture<>(); + createSystemTopicFactoryIfNeeded(); + SystemTopicClient systemTopicClient = namespaceEventsSystemTopicFactory + .createTopicPoliciesSystemTopicClient(namespace); + Backoff backoff = new Backoff(1, TimeUnit.SECONDS, 3, TimeUnit.SECONDS, 10, TimeUnit.SECONDS); + RetryUtil.retryAsynchronously(systemTopicClient::newReaderAsync, backoff, pulsarService.getExecutor(), result); + return result; + } + @Override public CompletableFuture removeOwnedNamespaceBundleAsync(NamespaceBundle namespaceBundle) { NamespaceName namespace = namespaceBundle.getNamespaceObject(); 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 new file mode 100644 index 0000000000000..8607cf5327ff8 --- /dev/null +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/util/RetryUtil.java @@ -0,0 +1,67 @@ +/** + * 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.util; + +import org.apache.pulsar.client.impl.Backoff; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; + +public class RetryUtil { + private static final Logger log = LoggerFactory.getLogger(RetryUtil.class); + + public static void retryAsynchronously(Supplier> supplier, Backoff backoff, + ScheduledExecutorService scheduledExecutorService, + CompletableFuture callback) { + if (backoff.getMax() <= 0) { + throw new IllegalArgumentException("Illegal max retry time"); + } + if (backoff.getInitial() <= 0) { + throw new IllegalArgumentException("Illegal initial time"); + } + scheduledExecutorService.execute(() -> + executeWithRetry(supplier, backoff, scheduledExecutorService, callback)); + } + + private static void executeWithRetry(Supplier> supplier, Backoff backoff, + ScheduledExecutorService scheduledExecutorService, + CompletableFuture callback) { + 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); + } + return; + } + callback.complete(result); + }); + } + +} diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/util/RetryUtilTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/util/RetryUtilTest.java new file mode 100644 index 0000000000000..1cfe2f8cb4f7d --- /dev/null +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/util/RetryUtilTest.java @@ -0,0 +1,85 @@ +/** + * 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.util; + +import org.apache.pulsar.client.impl.Backoff; +import org.apache.pulsar.client.impl.BackoffBuilder; +import org.apache.pulsar.common.util.FutureUtil; +import org.testng.annotations.Test; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; + +@Test(groups = "utils") +public class RetryUtilTest { + + + @Test + public void testFailAndRetry() throws Exception { + ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); + CompletableFuture callback = new CompletableFuture<>(); + AtomicInteger atomicInteger = new AtomicInteger(0); + Backoff backoff = new BackoffBuilder() + .setInitialTime(100, TimeUnit.MILLISECONDS) + .setMax(2000, TimeUnit.MILLISECONDS) + .setMandatoryStop(5000, TimeUnit.MILLISECONDS) + .create(); + RetryUtil.retryAsynchronously(() -> { + CompletableFuture future = new CompletableFuture<>(); + atomicInteger.incrementAndGet(); + if (atomicInteger.get() < 5) { + future.completeExceptionally(new RuntimeException("fail")); + } else { + future.complete(true); + } + return future; + }, backoff, executor, callback); + assertTrue(callback.get()); + assertEquals(atomicInteger.get(), 5); + executor.shutdownNow(); + } + + @Test + public void testFail() throws Exception { + ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); + CompletableFuture callback = new CompletableFuture<>(); + Backoff backoff = new BackoffBuilder() + .setInitialTime(500, TimeUnit.MILLISECONDS) + .setMax(2000, TimeUnit.MILLISECONDS) + .setMandatoryStop(5000, TimeUnit.MILLISECONDS) + .create(); + long start = System.currentTimeMillis(); + RetryUtil.retryAsynchronously(() -> + FutureUtil.failedFuture(new RuntimeException("fail")), backoff, executor, callback); + try { + callback.get(); + } catch (Exception e) { + assertTrue(e.getMessage().contains("fail")); + } + long time = System.currentTimeMillis() - start; + assertTrue(time >= 5000 - 2000, "Duration:" + time); + executor.shutdownNow(); + } +} From 957d41c730fa3993b07c127b65f360d403d09fe9 Mon Sep 17 00:00:00 2001 From: mattison chao Date: Tue, 15 Feb 2022 16:55:40 +0800 Subject: [PATCH 2/6] [Branch 2.7] Fix get topic policy timeout. --- .../service/SystemTopicBasedTopicPoliciesService.java | 6 ++++++ 1 file changed, 6 insertions(+) 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 4448b72dca654..b22e74dc93b83 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 @@ -273,6 +273,9 @@ private void initPolicesCache(SystemTopicClient.Reader reader, CompletableFuture reader.getSystemTopic().getTopicName(), ex); future.completeExceptionally(ex); readerCaches.remove(reader.getSystemTopic().getTopicName().getNamespaceObject()); + policyCacheInitMap.remove(reader.getSystemTopic().getTopicName().getNamespaceObject()); + reader.closeAsync(); + return; } if (hasMore) { reader.readNextAsync().whenComplete((msg, e) -> { @@ -281,6 +284,9 @@ private void initPolicesCache(SystemTopicClient.Reader reader, CompletableFuture reader.getSystemTopic().getTopicName(), ex); future.completeExceptionally(e); readerCaches.remove(reader.getSystemTopic().getTopicName().getNamespaceObject()); + policyCacheInitMap.remove(reader.getSystemTopic().getTopicName().getNamespaceObject()); + reader.closeAsync(); + return; } refreshTopicPoliciesCache(msg); if (log.isDebugEnabled()) { From d7871e270897928fadfae96c8e3ce4915fe969d5 Mon Sep 17 00:00:00 2001 From: mattison chao Date: Tue, 15 Feb 2022 20:33:48 +0800 Subject: [PATCH 3/6] Revert "[Branch 2.7] Fix get topic policy timeout." This reverts commit 957d41c730fa3993b07c127b65f360d403d09fe9. --- .../service/SystemTopicBasedTopicPoliciesService.java | 6 ------ 1 file changed, 6 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 b22e74dc93b83..4448b72dca654 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 @@ -273,9 +273,6 @@ private void initPolicesCache(SystemTopicClient.Reader reader, CompletableFuture reader.getSystemTopic().getTopicName(), ex); future.completeExceptionally(ex); readerCaches.remove(reader.getSystemTopic().getTopicName().getNamespaceObject()); - policyCacheInitMap.remove(reader.getSystemTopic().getTopicName().getNamespaceObject()); - reader.closeAsync(); - return; } if (hasMore) { reader.readNextAsync().whenComplete((msg, e) -> { @@ -284,9 +281,6 @@ private void initPolicesCache(SystemTopicClient.Reader reader, CompletableFuture reader.getSystemTopic().getTopicName(), ex); future.completeExceptionally(e); readerCaches.remove(reader.getSystemTopic().getTopicName().getNamespaceObject()); - policyCacheInitMap.remove(reader.getSystemTopic().getTopicName().getNamespaceObject()); - reader.closeAsync(); - return; } refreshTopicPoliciesCache(msg); if (log.isDebugEnabled()) { From 2efc3e2cabced3577e35efbd4b6deab4e9d3b90a Mon Sep 17 00:00:00 2001 From: mattison chao Date: Tue, 15 Feb 2022 20:33:58 +0800 Subject: [PATCH 4/6] Revert "[Branch 2.7] Fix get topic policy timeout." This reverts commit 209e4241c2df251c33fdbbbd5de0c22e50572f33. --- .../pulsar/broker/admin/AdminResource.java | 59 +++++++------ .../SystemTopicBasedTopicPoliciesService.java | 54 ++++-------- .../apache/pulsar/client/util/RetryUtil.java | 67 --------------- .../pulsar/client/util/RetryUtilTest.java | 85 ------------------- 4 files changed, 49 insertions(+), 216 deletions(-) delete mode 100644 pulsar-client/src/main/java/org/apache/pulsar/client/util/RetryUtil.java delete mode 100644 pulsar-client/src/test/java/org/apache/pulsar/client/util/RetryUtilTest.java diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java index 6fd4419bbac10..f0c028a8146ce 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java @@ -33,9 +33,10 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; -import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Collectors; import javax.servlet.ServletContext; import javax.ws.rs.WebApplicationException; @@ -43,6 +44,7 @@ import javax.ws.rs.core.Response; import javax.ws.rs.core.Response.Status; import javax.ws.rs.core.UriBuilder; + import com.google.errorprone.annotations.CanIgnoreReturnValue; import org.apache.bookkeeper.util.ZkUtils; import org.apache.pulsar.broker.PulsarService; @@ -54,7 +56,6 @@ import org.apache.pulsar.client.admin.internal.TopicsImpl; import org.apache.pulsar.client.impl.Backoff; import org.apache.pulsar.client.impl.BackoffBuilder; -import org.apache.pulsar.client.util.RetryUtil; import org.apache.pulsar.common.api.proto.PulsarApi; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.common.naming.Constants; @@ -551,36 +552,44 @@ protected BacklogQuota namespaceBacklogQuota(String namespace, String namespaceP } protected CompletableFuture> getTopicPoliciesAsyncWithRetry(TopicName topicName) { - try { - checkTopicLevelPolicyEnable(); - }catch (RestException ex) { - log.error("[{}] Failed to get topic policies {}", clientAppId(), topicName, ex); - return FutureUtil.failedFuture(ex); - } - return internalGetTopicPoliciesAsyncWithRetry(topicName, null, pulsar().getExecutor()); + return internalGetTopicPoliciesAsyncWithRetry(topicName, + new AtomicLong(DEFAULT_GET_TOPIC_POLICY_TIMEOUT), null, null); } protected CompletableFuture> internalGetTopicPoliciesAsyncWithRetry(TopicName topicName, + final AtomicLong remainingTime, final Backoff backoff, - ScheduledExecutorService scheduledExecutorService) { - CompletableFuture> response = new CompletableFuture<>(); - Backoff usedBackoff = backoff == null ? new BackoffBuilder() - .setInitialTime(500, TimeUnit.MILLISECONDS) - .setMandatoryStop(DEFAULT_GET_TOPIC_POLICY_TIMEOUT, TimeUnit.MILLISECONDS) - .setMax(DEFAULT_GET_TOPIC_POLICY_TIMEOUT, TimeUnit.MILLISECONDS) - .create() : backoff; + CompletableFuture> future) { + CompletableFuture> response = future == null ? new CompletableFuture<>() : future; try { - RetryUtil.retryAsynchronously(() -> { - CompletableFuture> future = new CompletableFuture<>(); - try { - future.complete(Optional.ofNullable(pulsar().getTopicPoliciesService() - .getTopicPolicies(topicName))); - } catch (BrokerServiceException.TopicPoliciesCacheNotInitException exception) { - future.completeExceptionally(exception); + checkTopicLevelPolicyEnable(); + response.complete(Optional.ofNullable(pulsar() + .getTopicPoliciesService().getTopicPolicies(topicName))); + } catch (RestException re) { + response.completeExceptionally(re); + } catch (BrokerServiceException.TopicPoliciesCacheNotInitException e) { + Backoff usedBackoff = backoff == null ? new BackoffBuilder() + .setInitialTime(500, TimeUnit.MILLISECONDS) + .setMandatoryStop(DEFAULT_GET_TOPIC_POLICY_TIMEOUT, TimeUnit.MILLISECONDS) + .setMax(DEFAULT_GET_TOPIC_POLICY_TIMEOUT, TimeUnit.MILLISECONDS) + .create() : backoff; + long nextDelay = Math.min(usedBackoff.next(), remainingTime.get()); + if (nextDelay <= 0) { + response.completeExceptionally(new TimeoutException( + String.format("Failed to get topic policy withing configured timeout %s ms", + DEFAULT_GET_TOPIC_POLICY_TIMEOUT))); + } else { + if (log.isDebugEnabled()) { + log.error("Topic {} policies have not been initialized yet, retry after {}ms", + topicName, nextDelay); } - return future; - }, usedBackoff, scheduledExecutorService, response); + pulsar().getExecutor().schedule(() -> { + remainingTime.addAndGet(-nextDelay); + internalGetTopicPoliciesAsyncWithRetry(topicName, remainingTime, usedBackoff, response); + }, nextDelay, TimeUnit.MILLISECONDS); + } } catch (Exception e) { + log.error("[{}] Failed to get topic policies {}", clientAppId(), topicName, e); response.completeExceptionally(e); } return response; 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 4448b72dca654..3dc5a1014cb67 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 @@ -25,7 +25,6 @@ import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.apache.pulsar.broker.PulsarServerException; import org.apache.pulsar.broker.PulsarService; @@ -35,8 +34,6 @@ import org.apache.pulsar.broker.systopic.SystemTopicClient; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.PulsarClientException; -import org.apache.pulsar.client.impl.Backoff; -import org.apache.pulsar.client.util.RetryUtil; import org.apache.pulsar.common.events.ActionType; import org.apache.pulsar.common.events.EventType; import org.apache.pulsar.common.events.PulsarEvent; @@ -147,10 +144,6 @@ public boolean cacheIsInitialized(TopicName topicName) { @Override public TopicPolicies getTopicPolicies(TopicName topicName) throws TopicPoliciesCacheNotInitException { - if (!policyCacheInitMap.containsKey(topicName.getNamespaceObject())) { - NamespaceName namespace = topicName.getNamespaceObject(); - prepareInitPoliciesCache(namespace, new CompletableFuture<>()); - } if (policyCacheInitMap.containsKey(topicName.getNamespaceObject()) && !policyCacheInitMap.get(topicName.getNamespaceObject())) { throw new TopicPoliciesCacheNotInitException(); @@ -158,25 +151,6 @@ public TopicPolicies getTopicPolicies(TopicName topicName) throws TopicPoliciesC return policiesCache.get(TopicName.get(topicName.getPartitionedTopicName())); } - private void prepareInitPoliciesCache(NamespaceName namespace, CompletableFuture result) { - if (policyCacheInitMap.putIfAbsent(namespace, false) == null) { - CompletableFuture readerCompletableFuture = - createSystemTopicClientWithRetry(namespace); - readerCaches.put(namespace, readerCompletableFuture); - readerCompletableFuture.whenComplete((reader, ex) -> { - if (ex != null) { - log.error("[{}] Failed to create reader on __change_events topic", namespace, ex); - result.completeExceptionally(ex); - readerCaches.remove(namespace); - reader.closeAsync(); - } else { - initPolicesCache(reader, result); - result.thenRun(() -> readMorePolicies(reader)); - } - }); - } - } - @Override public CompletableFuture getTopicPoliciesBypassCacheAsync(TopicName topicName) { CompletableFuture result = new CompletableFuture<>(); @@ -200,30 +174,32 @@ public CompletableFuture addOwnedNamespaceBundleAsync(NamespaceBundle name result.complete(null); return result; } + createSystemTopicFactoryIfNeeded(); synchronized (this) { if (readerCaches.get(namespace) != null) { ownedBundlesCountPerNamespace.get(namespace).incrementAndGet(); result.complete(null); - return result; } else { + SystemTopicClient systemTopicClient = namespaceEventsSystemTopicFactory.createSystemTopic(namespace + , EventType.TOPIC_POLICY); ownedBundlesCountPerNamespace.putIfAbsent(namespace, new AtomicInteger(1)); - prepareInitPoliciesCache(namespace, result); + policyCacheInitMap.put(namespace, false); + CompletableFuture readerCompletableFuture = systemTopicClient.newReaderAsync(); + readerCaches.put(namespace, readerCompletableFuture); + readerCompletableFuture.whenComplete((reader, ex) -> { + if (ex != null) { + log.error("[{}] Failed to create reader on __change_events topic", namespace, ex); + result.completeExceptionally(ex); + } else { + initPolicesCache(reader, result); + result.thenRun(() -> readMorePolicies(reader)); + } + }); } } return result; } - protected CompletableFuture createSystemTopicClientWithRetry( - NamespaceName namespace) { - CompletableFuture result = new CompletableFuture<>(); - createSystemTopicFactoryIfNeeded(); - SystemTopicClient systemTopicClient = namespaceEventsSystemTopicFactory - .createTopicPoliciesSystemTopicClient(namespace); - Backoff backoff = new Backoff(1, TimeUnit.SECONDS, 3, TimeUnit.SECONDS, 10, TimeUnit.SECONDS); - RetryUtil.retryAsynchronously(systemTopicClient::newReaderAsync, backoff, pulsarService.getExecutor(), result); - return result; - } - @Override public CompletableFuture removeOwnedNamespaceBundleAsync(NamespaceBundle namespaceBundle) { NamespaceName namespace = namespaceBundle.getNamespaceObject(); 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 deleted file mode 100644 index 8607cf5327ff8..0000000000000 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/util/RetryUtil.java +++ /dev/null @@ -1,67 +0,0 @@ -/** - * 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.util; - -import org.apache.pulsar.client.impl.Backoff; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import java.util.function.Supplier; - -public class RetryUtil { - private static final Logger log = LoggerFactory.getLogger(RetryUtil.class); - - public static void retryAsynchronously(Supplier> supplier, Backoff backoff, - ScheduledExecutorService scheduledExecutorService, - CompletableFuture callback) { - if (backoff.getMax() <= 0) { - throw new IllegalArgumentException("Illegal max retry time"); - } - if (backoff.getInitial() <= 0) { - throw new IllegalArgumentException("Illegal initial time"); - } - scheduledExecutorService.execute(() -> - executeWithRetry(supplier, backoff, scheduledExecutorService, callback)); - } - - private static void executeWithRetry(Supplier> supplier, Backoff backoff, - ScheduledExecutorService scheduledExecutorService, - CompletableFuture callback) { - 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); - } - return; - } - callback.complete(result); - }); - } - -} diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/util/RetryUtilTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/util/RetryUtilTest.java deleted file mode 100644 index 1cfe2f8cb4f7d..0000000000000 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/util/RetryUtilTest.java +++ /dev/null @@ -1,85 +0,0 @@ -/** - * 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.util; - -import org.apache.pulsar.client.impl.Backoff; -import org.apache.pulsar.client.impl.BackoffBuilder; -import org.apache.pulsar.common.util.FutureUtil; -import org.testng.annotations.Test; - -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; - -import static org.testng.Assert.assertEquals; -import static org.testng.Assert.assertTrue; - -@Test(groups = "utils") -public class RetryUtilTest { - - - @Test - public void testFailAndRetry() throws Exception { - ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); - CompletableFuture callback = new CompletableFuture<>(); - AtomicInteger atomicInteger = new AtomicInteger(0); - Backoff backoff = new BackoffBuilder() - .setInitialTime(100, TimeUnit.MILLISECONDS) - .setMax(2000, TimeUnit.MILLISECONDS) - .setMandatoryStop(5000, TimeUnit.MILLISECONDS) - .create(); - RetryUtil.retryAsynchronously(() -> { - CompletableFuture future = new CompletableFuture<>(); - atomicInteger.incrementAndGet(); - if (atomicInteger.get() < 5) { - future.completeExceptionally(new RuntimeException("fail")); - } else { - future.complete(true); - } - return future; - }, backoff, executor, callback); - assertTrue(callback.get()); - assertEquals(atomicInteger.get(), 5); - executor.shutdownNow(); - } - - @Test - public void testFail() throws Exception { - ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); - CompletableFuture callback = new CompletableFuture<>(); - Backoff backoff = new BackoffBuilder() - .setInitialTime(500, TimeUnit.MILLISECONDS) - .setMax(2000, TimeUnit.MILLISECONDS) - .setMandatoryStop(5000, TimeUnit.MILLISECONDS) - .create(); - long start = System.currentTimeMillis(); - RetryUtil.retryAsynchronously(() -> - FutureUtil.failedFuture(new RuntimeException("fail")), backoff, executor, callback); - try { - callback.get(); - } catch (Exception e) { - assertTrue(e.getMessage().contains("fail")); - } - long time = System.currentTimeMillis() - start; - assertTrue(time >= 5000 - 2000, "Duration:" + time); - executor.shutdownNow(); - } -} From aa268400b8df416c6c5c63b8c2dd373a2e88000e Mon Sep 17 00:00:00 2001 From: mattison chao Date: Tue, 15 Feb 2022 20:47:09 +0800 Subject: [PATCH 5/6] [Branch 2.7] Fix get topic policy timeout. --- .../SystemTopicBasedTopicPoliciesService.java | 46 +++++++++++++------ 1 file changed, 31 insertions(+), 15 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 3dc5a1014cb67..0541c7a498534 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 @@ -144,6 +144,10 @@ public boolean cacheIsInitialized(TopicName topicName) { @Override public TopicPolicies getTopicPolicies(TopicName topicName) throws TopicPoliciesCacheNotInitException { + if (!policyCacheInitMap.containsKey(topicName.getNamespaceObject())) { + NamespaceName namespace = topicName.getNamespaceObject(); + prepareInitPoliciesCache(namespace, new CompletableFuture<>()); + } if (policyCacheInitMap.containsKey(topicName.getNamespaceObject()) && !policyCacheInitMap.get(topicName.getNamespaceObject())) { throw new TopicPoliciesCacheNotInitException(); @@ -151,6 +155,25 @@ public TopicPolicies getTopicPolicies(TopicName topicName) throws TopicPoliciesC return policiesCache.get(TopicName.get(topicName.getPartitionedTopicName())); } + private void prepareInitPoliciesCache(NamespaceName namespace, CompletableFuture result) { + if (policyCacheInitMap.putIfAbsent(namespace, false) == null) { + CompletableFuture readerCompletableFuture = namespaceEventsSystemTopicFactory + .createTopicPoliciesSystemTopicClient(namespace).newReaderAsync(); + readerCaches.put(namespace, readerCompletableFuture); + readerCompletableFuture.whenComplete((reader, ex) -> { + if (ex != null) { + log.error("[{}] Failed to create reader on __change_events topic", namespace, ex); + result.completeExceptionally(ex); + readerCaches.remove(namespace); + reader.closeAsync(); + } else { + initPolicesCache(reader, result); + result.thenRun(() -> readMorePolicies(reader)); + } + }); + } + } + @Override public CompletableFuture getTopicPoliciesBypassCacheAsync(TopicName topicName) { CompletableFuture result = new CompletableFuture<>(); @@ -180,21 +203,8 @@ public CompletableFuture addOwnedNamespaceBundleAsync(NamespaceBundle name ownedBundlesCountPerNamespace.get(namespace).incrementAndGet(); result.complete(null); } else { - SystemTopicClient systemTopicClient = namespaceEventsSystemTopicFactory.createSystemTopic(namespace - , EventType.TOPIC_POLICY); ownedBundlesCountPerNamespace.putIfAbsent(namespace, new AtomicInteger(1)); - policyCacheInitMap.put(namespace, false); - CompletableFuture readerCompletableFuture = systemTopicClient.newReaderAsync(); - readerCaches.put(namespace, readerCompletableFuture); - readerCompletableFuture.whenComplete((reader, ex) -> { - if (ex != null) { - log.error("[{}] Failed to create reader on __change_events topic", namespace, ex); - result.completeExceptionally(ex); - } else { - initPolicesCache(reader, result); - result.thenRun(() -> readMorePolicies(reader)); - } - }); + prepareInitPoliciesCache(namespace, result); } } return result; @@ -249,14 +259,20 @@ private void initPolicesCache(SystemTopicClient.Reader reader, CompletableFuture reader.getSystemTopic().getTopicName(), ex); future.completeExceptionally(ex); readerCaches.remove(reader.getSystemTopic().getTopicName().getNamespaceObject()); + policyCacheInitMap.remove(reader.getSystemTopic().getTopicName().getNamespaceObject()); + reader.closeAsync(); + return; } if (hasMore) { reader.readNextAsync().whenComplete((msg, e) -> { if (e != null) { log.error("[{}] Failed to read event from the system topic.", - reader.getSystemTopic().getTopicName(), ex); + reader.getSystemTopic().getTopicName(), e); future.completeExceptionally(e); readerCaches.remove(reader.getSystemTopic().getTopicName().getNamespaceObject()); + policyCacheInitMap.remove(reader.getSystemTopic().getTopicName().getNamespaceObject()); + reader.closeAsync(); + return; } refreshTopicPoliciesCache(msg); if (log.isDebugEnabled()) { From b44968ccccfb4cd5b5da3d9f07d93cfb1a331072 Mon Sep 17 00:00:00 2001 From: mattison chao Date: Tue, 15 Feb 2022 21:02:39 +0800 Subject: [PATCH 6/6] Add test --- ...temTopicBasedTopicPoliciesServiceTest.java | 37 ++++++++++++++++++- 1 file changed, 36 insertions(+), 1 deletion(-) 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 9eb866980bfd1..c8068404e2868 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 @@ -22,6 +22,7 @@ import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; import org.apache.pulsar.broker.service.BrokerServiceException.TopicPoliciesCacheNotInitException; import org.apache.pulsar.broker.systopic.NamespaceEventsSystemTopicFactory; +import org.apache.pulsar.broker.systopic.SystemTopicClient; import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.common.naming.NamespaceName; import org.apache.pulsar.common.naming.TopicName; @@ -33,8 +34,13 @@ import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; - +import java.lang.reflect.Field; +import java.util.Map; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; public class SystemTopicBasedTopicPoliciesServiceTest extends MockedPulsarServiceBaseTest { @@ -193,4 +199,33 @@ private void prepareData() throws PulsarAdminException { systemTopicFactory = new NamespaceEventsSystemTopicFactory(pulsarClient); systemTopicBasedTopicPoliciesService = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService(); } + + @Test + public void testGetTopicPoliciesWithRetry() throws Exception { + Field initMapField = SystemTopicBasedTopicPoliciesService.class.getDeclaredField("policyCacheInitMap"); + initMapField.setAccessible(true); + Map initMap = (Map)initMapField.get(systemTopicBasedTopicPoliciesService); + initMap.remove(NamespaceName.get(NAMESPACE1)); + Field readerCaches = SystemTopicBasedTopicPoliciesService.class.getDeclaredField("readerCaches"); + readerCaches.setAccessible(true); + Map> readers = (Map)readerCaches.get(systemTopicBasedTopicPoliciesService); + readers.remove(NamespaceName.get(NAMESPACE1)); + TopicPolicies initPolicy = TopicPolicies.builder() + .maxConsumerPerTopic(10) + .build(); + ScheduledExecutorService executors = Executors.newScheduledThreadPool(1); + executors.schedule(() -> { + try { + systemTopicBasedTopicPoliciesService.updateTopicPoliciesAsync(TOPIC1, initPolicy).get(); + } catch (Exception ignore) {} + }, 2000, TimeUnit.MILLISECONDS); + Awaitility.await().untilAsserted(() -> { + try { + TopicPolicies topicPolicies = systemTopicBasedTopicPoliciesService.getTopicPolicies(TOPIC1); + Assert.assertNotNull(topicPolicies); + } catch (Exception ex) { + Assert.fail(); + } + }); + } }