opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-24 16:04:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #4734]Fix DLedgerController startScheduling concurrency problem (#4735)

Browse Source
This commit is contained in:
mxsm
2022-07-30 20:54:06 +08:00
committed by GitHub
parent 975c232f3e
commit c489a70baa
1 changed files with 4 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
controller/src/main/java/org/apache/rocketmq/controller/impl/DLedgerController.java
+4 -5
View File
@@ -33,6 +33,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiPredicate;
import java.util.function.Supplier;
import org.apache.rocketmq.common.ServiceThread;
@@ -76,7 +77,7 @@ public class DLedgerController implements Controller {
private final DLedgerControllerStateMachine statemachine;
// Usr for checking whether the broker is alive
private BiPredicate<String, String> brokerAlivePredicate;
private volatile boolean isScheduling = false;
private AtomicBoolean isScheduling = new AtomicBoolean(false);
public DLedgerController(final ControllerConfig config, final BiPredicate<String, String> brokerAlivePredicate) {
this(config, brokerAlivePredicate, null, null, null);
@@ -119,18 +120,16 @@ public class DLedgerController implements Controller {
@Override
public void startScheduling() {
if (!this.isScheduling) {
if (this.isScheduling.compareAndSet(false, true)) {
log.info("Start scheduling controller events");
this.isScheduling = true;
this.scheduler.start();
}
}
@Override
public void stopScheduling() {
if (this.isScheduling) {
if (this.isScheduling.compareAndSet(true, false)) {
log.info("Stop scheduling controller events");
this.isScheduling = false;
this.scheduler.shutdown(true);
}
}