From 7c5f9b0727d5f08e6e9160e9482db0d1d616fed9 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Wed, 7 Jun 2023 16:39:45 +0300 Subject: [PATCH] [fix][broker] Restore solution for certain topic unloading race conditions (#20527) (cherry picked from commit 03f916702ec2a887833543fdea5a2d5305a87302) --- .../java/org/apache/pulsar/broker/service/BrokerService.java | 4 ---- .../pulsar/broker/service/persistent/PersistentTopic.java | 2 +- .../apache/pulsar/broker/service/BrokerBkEnsemblesTests.java | 4 ++-- 3 files changed, 3 insertions(+), 7 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java index efe7235802fb6..263ee1fde2292 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java @@ -1909,10 +1909,6 @@ public AuthorizationService getAuthorizationService() { return authorizationService; } - public CompletableFuture removeTopicFromCache(String topicName) { - return removeTopicFutureFromCache(topicName, null); - } - public CompletableFuture removeTopicFromCache(Topic topic) { Optional>> createTopicFuture = findTopicFutureInCache(topic); if (!createTopicFuture.isPresent()){ diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index c502e8c107372..b807f218db725 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -1326,7 +1326,7 @@ public void closeFailed(ManagedLedgerException exception, Object ctx) { } private void disposeTopic(CompletableFuture closeFuture) { - brokerService.removeTopicFromCache(topic) + brokerService.removeTopicFromCache(PersistentTopic.this) .thenRun(() -> { replicatedSubscriptionsController.ifPresent(ReplicatedSubscriptionsController::close); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java index aa63b224a9d1e..b3530dbc4446f 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java @@ -119,7 +119,7 @@ public void testCrashBrokerWithoutCursorLedgerLeak() throws Exception { // (3) remove topic and managed-ledger from broker which means topic is not closed gracefully consumer.close(); producer.close(); - pulsar.getBrokerService().removeTopicFromCache(topic1); + pulsar.getBrokerService().removeTopicFromCache(topic); ManagedLedgerFactoryImpl factory = (ManagedLedgerFactoryImpl) pulsar.getManagedLedgerFactory(); Field field = ManagedLedgerFactoryImpl.class.getDeclaredField("ledgers"); field.setAccessible(true); @@ -242,7 +242,7 @@ public void testSkipCorruptDataLedger() throws Exception { // clean managed-ledger and recreate topic to clean any data from the cache producer.close(); - pulsar.getBrokerService().removeTopicFromCache(topic1); + pulsar.getBrokerService().removeTopicFromCache(topic); ManagedLedgerFactoryImpl factory = (ManagedLedgerFactoryImpl) pulsar.getManagedLedgerFactory(); Field field = ManagedLedgerFactoryImpl.class.getDeclaredField("ledgers"); field.setAccessible(true);