You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by ji...@apache.org on 2022/09/17 10:03:57 UTC
[pulsar] branch branch-2.10 updated: [improve][broker] Cancel the loadShedding task when closing pulsar service (#17632)
This is an automated email from the ASF dual-hosted git repository.
jianghaiting pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.10 by this push:
new 9a1f483befe [improve][broker] Cancel the loadShedding task when closing pulsar service (#17632)
9a1f483befe is described below
commit 9a1f483befe565395489e877f51f7c018502a807
Author: ran <ga...@126.com>
AuthorDate: Thu Sep 15 15:52:10 2022 +0800
[improve][broker] Cancel the loadShedding task when closing pulsar service (#17632)
(cherry picked from commit cbbcd41cfc80793cf025ef98390339395ecd48ac)
---
.../org/apache/pulsar/broker/PulsarService.java | 5 +-
.../broker/loadbalance/LoadSheddingTask.java | 23 ++++---
.../pulsar/broker/PulsarServiceCloseTest.java | 79 ++++++++++++++++++++++
3 files changed, 97 insertions(+), 10 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index d50be2954eb..2a092b5b769 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -474,7 +474,10 @@ public class PulsarService implements AutoCloseable, ShutdownService {
this.leaderElectionService = null;
}
- // shutdown loadmanager before shutting down the broker
+ // cancel loadShedding task and shutdown the loadManager executor before shutting down the broker
+ if (this.loadSheddingTask != null) {
+ this.loadSheddingTask.cancel();
+ }
executorServicesShutdown.shutdown(loadManagerExecutor);
if (adminClient != null) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadSheddingTask.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadSheddingTask.java
index 73139ac4989..b4a01d89e52 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadSheddingTask.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadSheddingTask.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.broker.loadbalance;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.pulsar.broker.ServiceConfiguration;
@@ -37,6 +38,8 @@ public class LoadSheddingTask implements Runnable {
private volatile boolean isCancel = false;
+ private volatile ScheduledFuture<?> future;
+
public LoadSheddingTask(AtomicReference<LoadManager> loadManager,
ScheduledExecutorService loadManagerExecutor,
ServiceConfiguration config) {
@@ -47,24 +50,22 @@ public class LoadSheddingTask implements Runnable {
@Override
public void run() {
+ if (isCancel) {
+ return;
+ }
try {
loadManager.get().doLoadShedding();
} catch (Exception e) {
LOG.warn("Error during the load shedding", e);
} finally {
- if (!isCancel && loadManagerExecutor != null && config != null) {
- loadManagerExecutor.schedule(
- new LoadSheddingTask(loadManager, loadManagerExecutor, config),
- config.getLoadBalancerSheddingIntervalMinutes(),
- TimeUnit.MINUTES);
- }
+ start();
}
}
public void start() {
- if (loadManagerExecutor != null && config != null) {
- loadManagerExecutor.schedule(
- new LoadSheddingTask(loadManager, loadManagerExecutor, config),
+ if (!isCancel && loadManagerExecutor != null && config != null) {
+ future = loadManagerExecutor.schedule(
+ this,
config.getLoadBalancerSheddingIntervalMinutes(),
TimeUnit.MINUTES);
}
@@ -72,5 +73,9 @@ public class LoadSheddingTask implements Runnable {
public void cancel() {
isCancel = true;
+ if (future != null) {
+ future.cancel(false);
+ }
}
+
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/PulsarServiceCloseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/PulsarServiceCloseTest.java
new file mode 100644
index 00000000000..c424132855b
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/PulsarServiceCloseTest.java
@@ -0,0 +1,79 @@
+/**
+ * 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;
+
+import static org.mockito.Mockito.spy;
+import static org.testng.AssertJUnit.assertFalse;
+import static org.testng.AssertJUnit.assertTrue;
+
+import java.util.concurrent.ScheduledFuture;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.broker.loadbalance.LoadSheddingTask;
+import org.awaitility.reflect.WhiteboxImpl;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+@Test(groups = "broker")
+public class PulsarServiceCloseTest extends MockedPulsarServiceBaseTest {
+
+ @BeforeClass
+ @Override
+ protected void setup() throws Exception {
+ super.internalSetup();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ protected PulsarService startBrokerWithoutAuthorization(ServiceConfiguration conf) throws Exception {
+ conf.setBrokerShutdownTimeoutMs(1000 * 60 * 5);
+ conf.setLoadBalancerSheddingIntervalMinutes(30);
+ PulsarService pulsar = spy(newPulsarService(conf));
+ setupBrokerMocks(pulsar);
+ beforePulsarStartMocks(pulsar);
+ pulsar.start();
+ log.info("Pulsar started. brokerServiceUrl: {} webServiceAddress: {}", pulsar.getBrokerServiceUrl(),
+ pulsar.getWebServiceAddress());
+ return pulsar;
+ }
+
+ @Test(timeOut = 30_000)
+ public void closeInTimeTest() throws Exception {
+ LoadSheddingTask task = pulsar.getLoadSheddingTask();
+ boolean isCancel = WhiteboxImpl.getInternalState(task, "isCancel");
+ assertFalse(isCancel);
+ ScheduledFuture<?> loadSheddingFuture = WhiteboxImpl.getInternalState(task, "future");
+ assertFalse(loadSheddingFuture.isCancelled());
+
+ // The pulsar service is not used, so it should be closed gracefully in short time.
+ pulsar.close();
+
+ isCancel = WhiteboxImpl.getInternalState(task, "isCancel");
+ assertTrue(isCancel);
+ loadSheddingFuture = WhiteboxImpl.getInternalState(task, "future");
+ assertTrue(loadSheddingFuture.isCancelled());
+ }
+
+}