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

[ISSUE #6706] brokerPermission cannot stop messages flow in unwritable brokers

Browse Source
This commit is contained in:
Ji Juntao
2023-05-06 16:49:55 +08:00
committed by GitHub
parent 6770f92819
commit e73a9aa8a7
5 changed files with 18 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
+1 -1
View File
@@ -765,7 +765,7 @@ public class BrokerController {
}
if (this.brokerConfig.isEnableControllerMode()) {
this.replicasManager.setIsolatedAndBrokerPermission(false);
this.replicasManager.setFenced(true);
}
if (messageStore != null) {
broker/src/main/java/org/apache/rocketmq/broker/controller/ReplicasManager.java
+4 -16
View File
@@ -93,8 +93,6 @@ public class ReplicasManager {
private Long masterBrokerId;
private volatile int originalBrokerPermission = 0;
private BrokerMetadata brokerMetadata;
private TempBrokerMetadata tempBrokerMetadata;
@@ -203,7 +201,7 @@ public class ReplicasManager {
if (this.masterBrokerId != null || brokerElect()) {
LOGGER.info("Master in this broker set is elected, masterBrokerId: {}, masterBrokerAddr: {}", this.masterBrokerId, this.masterAddress);
this.state = State.RUNNING;
setIsolatedAndBrokerPermission(true);
setFenced(false);
LOGGER.info("All register process has been done, change state to: {}", this.state);
} else {
return false;
@@ -241,7 +239,6 @@ public class ReplicasManager {
synchronized (this) {
if (newMasterEpoch > this.masterEpoch) {
LOGGER.info("Begin to change to master, brokerName:{}, replicas:{}, new Epoch:{}", this.brokerConfig.getBrokerName(), this.brokerAddress, newMasterEpoch);
this.masterEpoch = newMasterEpoch;
if (this.masterBrokerId != null && this.masterBrokerId.equals(this.brokerControllerId) && this.brokerController.getBrokerConfig().getBrokerId() == MixAll.MASTER_ID) {
// Change SyncStateSet
@@ -873,17 +870,8 @@ public class ReplicasManager {
return tempBrokerMetadata;
}
public void setIsolatedAndBrokerPermission(boolean isBrokerRoleConfirmed) {
if (isBrokerRoleConfirmed) {
this.brokerController.setIsolated(false);
this.brokerConfig.setBrokerPermission(this.originalBrokerPermission);
this.brokerController.getMessageStore().getRunningFlags().makeIsolated(false);
} else {
// prohibit writing and reading before confirming the broker role
this.brokerController.setIsolated(true);
this.originalBrokerPermission = this.brokerConfig.getBrokerPermission();
this.brokerConfig.setBrokerPermission(0);
this.brokerController.getMessageStore().getRunningFlags().makeIsolated(true);
}
public void setFenced(boolean fenced) {
this.brokerController.setIsolated(fenced);
this.brokerController.getMessageStore().getRunningFlags().makeFenced(fenced);
}
}
store/src/main/java/org/apache/rocketmq/store/CommitLog.java
+1 -1
View File
@@ -555,7 +555,7 @@ public class CommitLog implements Swappable {
public long getConfirmOffset() {
if (this.defaultMessageStore.getBrokerConfig().isEnableControllerMode()) {
if (this.defaultMessageStore.getMessageStoreConfig().getBrokerRole() != BrokerRole.SLAVE && !this.defaultMessageStore.getRunningFlags().isIsolated()) {
if (this.defaultMessageStore.getMessageStoreConfig().getBrokerRole() != BrokerRole.SLAVE && !this.defaultMessageStore.getRunningFlags().isFenced()) {
if (((AutoSwitchHAService) this.defaultMessageStore.getHaService()).getLocalSyncStateSet().size() == 1) {
return this.defaultMessageStore.getMaxPhyOffset();
}
store/src/main/java/org/apache/rocketmq/store/RunningFlags.java
+8 -8
View File
@@ -28,7 +28,7 @@ public class RunningFlags {
private static final int DISK_FULL_BIT = 1 << 4;
private static final int ISOLATED_BIT = 1 << 5;
private static final int FENCED_BIT = 1 << 5;
private volatile int flagBits = 0;
@@ -51,8 +51,8 @@ public class RunningFlags {
return (this.flagBits & NOT_READABLE_BIT) == 0;
}
public boolean isIsolated() {
return (this.flagBits & ISOLATED_BIT) != 0;
public boolean isFenced() {
return (this.flagBits & FENCED_BIT) != 0;
}
public boolean getAndMakeNotReadable() {
@@ -72,7 +72,7 @@ public class RunningFlags {
}
public boolean isWriteable() {
if ((this.flagBits & (NOT_WRITEABLE_BIT | WRITE_LOGICS_QUEUE_ERROR_BIT | DISK_FULL_BIT | WRITE_INDEX_FILE_ERROR_BIT)) == 0) {
if ((this.flagBits & (NOT_WRITEABLE_BIT | WRITE_LOGICS_QUEUE_ERROR_BIT | DISK_FULL_BIT | WRITE_INDEX_FILE_ERROR_BIT | FENCED_BIT)) == 0) {
return true;
}
@@ -100,11 +100,11 @@ public class RunningFlags {
this.flagBits |= WRITE_LOGICS_QUEUE_ERROR_BIT;
}
public void makeIsolated(boolean isolated) {
if (isolated) {
this.flagBits |= ISOLATED_BIT;
public void makeFenced(boolean fenced) {
if (fenced) {
this.flagBits |= FENCED_BIT;
} else {
this.flagBits &= ~ISOLATED_BIT;
this.flagBits &= ~FENCED_BIT;
}
}
store/src/test/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHATest.java
+4 -4
View File
@@ -232,11 +232,11 @@ public class AutoSwitchHATest {
storeConfig2.setBrokerRole(BrokerRole.SYNC_MASTER);
messageStore2 = buildMessageStore(storeConfig2, 2L);
messageStore2.getRunningFlags().makeIsolated(true);
messageStore2.getRunningFlags().makeFenced(true);
assertTrue(messageStore2.load());
messageStore2.start();
messageStore2.getHaService().changeToMaster(2);
messageStore2.getRunningFlags().makeIsolated(false);
messageStore2.getRunningFlags().makeFenced(false);
((AutoSwitchHAService) messageStore2.getHaService()).setSyncStateSet(new HashSet<>(Collections.singletonList(2L)));
// Put message on master
@@ -493,10 +493,10 @@ public class AutoSwitchHATest {
storeCheckpoint.setConfirmPhyOffset(setConfirmOffset);
storeCheckpoint.shutdown();
messageStore2 = buildMessageStore(storeConfig2, 2L);
messageStore2.getRunningFlags().makeIsolated(true);
messageStore2.getRunningFlags().makeFenced(true);
assertTrue(messageStore2.load());
messageStore2.start();
messageStore2.getRunningFlags().makeIsolated(false);
messageStore2.getRunningFlags().makeFenced(false);
assertEquals(setConfirmOffset, messageStore2.getConfirmOffset());
checkMessage(this.messageStore2, 5, 0);
}