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

[Summer of code] Fix some bugs in controller mode.

Browse Source 
* modify pom for using dledger

* rename option

* 1.send heartbeat to haclient when get epochEntry failed to hold connection.
2.update lastCatchupTimeMs in time.

* fix some bugs
This commit is contained in:
hzh0425
2022-05-26 20:44:34 +08:00
committed by GitHub
parent 25ee6eb322
commit 97e6cecbf9
13 changed files with 64 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
+3 -3
View File
@@ -303,7 +303,7 @@ public class BrokerConfig extends BrokerIdentity {
/**
* Whether the controller is deployed independently
*/
private boolean isControllerDeployedStandAlone = false;
private boolean controllerDeployedStandAlone = false;
/**
* If isControllerDeployedStandAlone = false, controllerAddr should be the addresses of all name-srv which running the controller instance.
@@ -1302,11 +1302,11 @@ public class BrokerConfig extends BrokerIdentity {
}
public boolean isControllerDeployedStandAlone() {
return isControllerDeployedStandAlone;
return controllerDeployedStandAlone;
}
public void setControllerDeployedStandAlone(boolean controllerDeployedStandAlone) {
isControllerDeployedStandAlone = controllerDeployedStandAlone;
this.controllerDeployedStandAlone = controllerDeployedStandAlone;
}
public String getControllerAddr() {
common/src/main/java/org/apache/rocketmq/common/namesrv/ControllerConfig.java
+5 -5
View File
@@ -27,7 +27,7 @@ public class ControllerConfig {
/**
* Is startup the controller in this name-srv
*/
private boolean isStartupController = false;
private boolean enableStartupController = false;
/**
* Interval of periodic scanning for non-active broker;
@@ -76,12 +76,12 @@ public class ControllerConfig {
this.configStorePath = configStorePath;
}
public boolean isStartupController() {
return isStartupController;
public boolean isEnableStartupController() {
return enableStartupController;
}
public void setStartupController(boolean startupController) {
isStartupController = startupController;
public void setEnableStartupController(boolean enableStartupController) {
this.enableStartupController = enableStartupController;
}
public long getScanNotActiveBrokerInterval() {
controller/pom.xml
+10 -1
View File
@@ -30,7 +30,16 @@
<dependency>
<groupId>io.openmessaging.storage</groupId>
<artifactId>dledger</artifactId>
<version>0.2.5</version>
<exclusions>
<exclusion>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-remoting</artifactId>
</exclusion>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
namesrv/pom.xml
-5
View File
@@ -28,11 +28,6 @@
<name>rocketmq-namesrv ${project.version}</name>
<dependencies>
<dependency>
<groupId>io.openmessaging.storage</groupId>
<artifactId>dledger</artifactId>
<version>0.2.5</version>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>rocketmq-controller</artifactId>
namesrv/src/main/java/org/apache/rocketmq/namesrv/NamesrvController.java
+4 -4
View File
@@ -110,7 +110,7 @@ public class NamesrvController {
this.kvConfigManager = new KVConfigManager(this);
this.brokerHousekeepingService = new BrokerHousekeepingService(this);
this.routeInfoManager = new RouteInfoManager(namesrvConfig, this);
if (controllerConfig.isStartupController()) {
if (controllerConfig.isEnableStartupController()) {
this.controller = new DledgerController(controllerConfig, this.routeInfoManager::isBrokerAlive);
this.routeInfoManager.setController(this.controller);
}
@@ -155,7 +155,7 @@ public class NamesrvController {
}
};
if (this.controllerConfig.isStartupController()) {
if (this.controllerConfig.isEnableStartupController()) {
this.controllerRequestThreadPoolQueue = new LinkedBlockingQueue<>(this.controllerConfig.getControllerRequestThreadPoolQueueCapacity());
this.controllerRequestExecutor = new ThreadPoolExecutor(
this.controllerConfig.getControllerThreadPoolNums(),
@@ -269,7 +269,7 @@ public class NamesrvController {
this.remotingServer.registerDefaultProcessor(new DefaultRequestProcessor(this), this.defaultExecutor);
if (controllerConfig.isStartupController()) {
if (controllerConfig.isEnableStartupController()) {
final ControllerRequestProcessor controllerRequestProcessor = new ControllerRequestProcessor(this.controller);
this.remotingServer.registerProcessor(RequestCode.CONTROLLER_ALTER_SYNC_STATE_SET, controllerRequestProcessor, this.controllerRequestExecutor);
this.remotingServer.registerProcessor(RequestCode.CONTROLLER_ELECT_MASTER, controllerRequestProcessor, this.controllerRequestExecutor);
@@ -290,7 +290,7 @@ public class NamesrvController {
this.routeInfoManager.start();
if (this.controllerConfig.isStartupController()) {
if (this.controllerConfig.isEnableStartupController()) {
this.controller.startup();
}
}
namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java
+1 -1
View File
@@ -600,7 +600,7 @@ public class RouteInfoManager {
}
// Check whether we need to elect a new master
if (this.namesrvController != null && this.namesrvController.getControllerConfig().isStartupController() && this.controller != null) {
if (this.namesrvController != null && this.namesrvController.getControllerConfig().isEnableStartupController() && this.controller != null) {
if (unRegisterRequest.getBrokerId() == 0) {
this.controller.electMaster(new ElectMasterRequestHeader(unRegisterRequest.getBrokerName()));
}
pom.xml
+5
View File
@@ -534,6 +534,11 @@
<artifactId>rocketmq-example</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>io.openmessaging.storage</groupId>
<artifactId>dledger</artifactId>
<version>0.2.5</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
store/pom.xml
-1
View File
@@ -31,7 +31,6 @@
<dependency>
<groupId>io.openmessaging.storage</groupId>
<artifactId>dledger</artifactId>
<version>0.2.3</version>
<exclusions>
<exclusion>
<groupId>org.apache.rocketmq</groupId>
store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
+4 -4
View File
@@ -36,7 +36,7 @@ public class MessageStoreConfig {
//The directory in which the commitlog is kept
@ImportantField
private String storePathEpochFile = System.getProperty("user.home") + File.separator + "store"
+ File.separator + "commitlog" + File.separator + "epochFileCheckpoint";
+ File.separator + "epochFileCheckpoint";
private String readOnlyCommitLogStorePaths = null;
@@ -306,7 +306,7 @@ public class MessageStoreConfig {
*/
private boolean syncFromLastFile = false;
private boolean isAsyncLearner = false;
private boolean asyncLearner = false;
public boolean isDebugLockEnable() {
return debugLockEnable;
@@ -1326,10 +1326,10 @@ public class MessageStoreConfig {
}
public boolean isAsyncLearner() {
return isAsyncLearner;
return asyncLearner;
}
public void setAsyncLearner(boolean asyncLearner) {
this.isAsyncLearner = asyncLearner;
this.asyncLearner = asyncLearner;
}
}
store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAClient.java
+6 -11
View File
@@ -156,14 +156,14 @@ public class AutoSwitchHAClient extends ServiceThread implements HAClient {
@Override public void updateMasterAddress(String newAddress) {
String currentAddr = this.masterAddress.get();
if (masterAddress.compareAndSet(currentAddr, newAddress)) {
if (!StringUtils.equals(newAddress, currentAddr) && masterAddress.compareAndSet(currentAddr, newAddress)) {
LOGGER.info("update master address, OLD: " + currentAddr + " NEW: " + newAddress);
}
}
@Override public void updateHaMasterAddress(String newAddress) {
String currentAddr = this.masterHaAddress.get();
if (masterHaAddress.compareAndSet(currentAddr, newAddress)) {
if (!StringUtils.equals(newAddress, currentAddr) && masterHaAddress.compareAndSet(currentAddr, newAddress)) {
LOGGER.info("update master ha address, OLD: " + currentAddr + " NEW: " + newAddress);
wakeup();
}
@@ -253,10 +253,10 @@ public class AutoSwitchHAClient extends ServiceThread implements HAClient {
// Original state
this.handshakeHeaderBuffer.putInt(HAConnectionState.HANDSHAKE.ordinal());
// IsSyncFromLastFile
short isSyncFromLastFile = this.haService.getDefaultMessageStore().getMessageStoreConfig().isSyncFromLastFile() ? (short)1 : (short) 0;
short isSyncFromLastFile = this.haService.getDefaultMessageStore().getMessageStoreConfig().isSyncFromLastFile() ? (short) 1 : (short) 0;
this.handshakeHeaderBuffer.putShort(isSyncFromLastFile);
// IsAsyncLearner role
short isAsyncLearner = this.haService.getDefaultMessageStore().getMessageStoreConfig().isAsyncLearner() ? (short)1 : (short) 0;
short isAsyncLearner = this.haService.getDefaultMessageStore().getMessageStoreConfig().isAsyncLearner() ? (short) 1 : (short) 0;
this.handshakeHeaderBuffer.putShort(isAsyncLearner);
// Address length
this.handshakeHeaderBuffer.putInt(this.localAddress == null ? 0 : this.localAddress.length());
@@ -437,7 +437,7 @@ public class AutoSwitchHAClient extends ServiceThread implements HAClient {
while (true) {
int diff = byteBufferRead.position() - AutoSwitchHAClient.this.processPosition;
if (diff >= AutoSwitchHAConnection.MSG_HEADER_SIZE) {
int processPosition = AutoSwitchHAClient.this.processPosition;
int processPosition = AutoSwitchHAClient.this.processPosition;
int masterState = byteBufferRead.getInt(processPosition + AutoSwitchHAConnection.MSG_HEADER_SIZE - 36);
int bodySize = byteBufferRead.getInt(processPosition + AutoSwitchHAConnection.MSG_HEADER_SIZE - 32);
long masterOffset = byteBufferRead.getLong(processPosition + AutoSwitchHAConnection.MSG_HEADER_SIZE - 28);
@@ -452,8 +452,6 @@ public class AutoSwitchHAClient extends ServiceThread implements HAClient {
masterState, AutoSwitchHAClient.this.currentState, bodySize, masterOffset, masterEpoch, masterEpochStartOffset, confirmOffset);
return true;
}
LOGGER.info("Receive master msg, masterState:{}, bodySize:{}, offset:{}, masterEpoch:{}, masterEpochStartOffset:{}, confirmOffset:{}",
HAConnectionState.values()[masterState], bodySize, masterOffset, masterEpoch, masterEpochStartOffset, confirmOffset);
if (diff >= (AutoSwitchHAConnection.MSG_HEADER_SIZE + bodySize)) {
switch (AutoSwitchHAClient.this.currentState) {
@@ -501,10 +499,7 @@ public class AutoSwitchHAClient extends ServiceThread implements HAClient {
AutoSwitchHAClient.this.confirmOffset = Math.min(confirmOffset, messageStore.getMaxPhyOffset());
if (bodySize > 0) {
final DefaultMessageStore messageStore = AutoSwitchHAClient.this.messageStore;
if (messageStore.appendToCommitLog(masterOffset, bodyData, 0, bodyData.length)) {
LOGGER.info("Slave append master log success, from {}, size {}, epoch:{}", masterOffset, bodySize, masterEpoch);
}
AutoSwitchHAClient.this.messageStore.appendToCommitLog(masterOffset, bodyData, 0, bodyData.length);
}
if (!reportSlaveMaxOffset()) {
store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAConnection.java
+21 -11
View File
@@ -503,10 +503,14 @@ public class AutoSwitchHAConnection implements HAConnection {
// Normal transfer method
private void buildTransferHeaderBuffer(long nextOffset, int bodySize) {
final EpochEntry entry = AutoSwitchHAConnection.this.epochCache.getEntry(AutoSwitchHAConnection.this.currentTransferEpoch);
EpochEntry entry = AutoSwitchHAConnection.this.epochCache.getEntry(AutoSwitchHAConnection.this.currentTransferEpoch);
if (entry == null) {
LOGGER.error("Failed to find epochEntry with epoch {} when build msg header", AutoSwitchHAConnection.this.currentTransferEpoch);
return;
if (bodySize > 0) {
return;
}
// Maybe it's used for heartbeat
entry = AutoSwitchHAConnection.this.epochCache.firstEntry();
}
// Build Header
this.byteBufferHeader.position(0);
@@ -529,20 +533,23 @@ public class AutoSwitchHAConnection implements HAConnection {
currentState, bodySize, nextOffset, entry.getEpoch(), entry.getStartOffset(), confirmOffset);
}
private boolean sendHeartbeatIfNeeded() throws Exception {
long interval = haService.getDefaultMessageStore().getSystemClock().now() - this.lastWriteTimestamp;
if (interval > haService.getDefaultMessageStore().getMessageStoreConfig().getHaSendHeartbeatInterval()) {
buildTransferHeaderBuffer(this.nextTransferFromWhere, 0);
return this.transferData(0);
}
return true;
}
private void transferToSlave() throws Exception {
if (this.lastWriteOver) {
long interval =
haService.getDefaultMessageStore().getSystemClock().now() - this.lastWriteTimestamp;
if (interval > haService.getDefaultMessageStore().getMessageStoreConfig()
.getHaSendHeartbeatInterval()) {
buildTransferHeaderBuffer(this.nextTransferFromWhere, 0);
this.lastWriteOver = this.transferData(0);
if (!this.lastWriteOver) {
return;
}
this.lastWriteOver = sendHeartbeatIfNeeded();
if (!this.lastWriteOver) {
return;
}
} else {
// maxTransferSize == -1 means to continue transfer remaining data.
@@ -596,6 +603,8 @@ public class AutoSwitchHAConnection implements HAConnection {
this.lastWriteOver = this.transferData(size);
} else {
// If size == 0, we should update the lastCatchupTimeMs
AutoSwitchHAConnection.this.lastCatchUpTimeMs = System.currentTimeMillis();
haService.getWaitNotifyObject().allWaitForRunning(100);
}
}
@@ -657,6 +666,7 @@ public class AutoSwitchHAConnection implements HAConnection {
EpochEntry epochEntry = AutoSwitchHAConnection.this.epochCache.findEpochEntryByOffset(this.nextTransferFromWhere);
if (epochEntry == null) {
LOGGER.error("Failed to find an epochEntry to match slaveRequestOffset {}", this.nextTransferFromWhere);
sendHeartbeatIfNeeded();
waitForRunning(500);
break;
}
store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAService.java
+4 -5
View File
@@ -35,7 +35,6 @@ import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.DispatchRequest;
import org.apache.rocketmq.store.SelectMappedBufferResult;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.ha.DefaultHAService;
import org.apache.rocketmq.store.ha.GroupTransferService;
import org.apache.rocketmq.store.ha.HAClient;
@@ -64,9 +63,6 @@ public class AutoSwitchHAService extends DefaultHAService {
this.defaultMessageStore = defaultMessageStore;
this.acceptSocketService = new AutoSwitchAcceptSocketService(defaultMessageStore.getMessageStoreConfig().getHaListenPort());
this.groupTransferService = new GroupTransferService(this, defaultMessageStore);
if (this.defaultMessageStore.getMessageStoreConfig().getBrokerRole() == BrokerRole.SLAVE) {
this.haClient = new AutoSwitchHAClient(this, defaultMessageStore, this.epochCache);
}
this.haConnectionStateNotificationService = new HAConnectionStateNotificationService(this, defaultMessageStore);
}
@@ -80,7 +76,7 @@ public class AutoSwitchHAService extends DefaultHAService {
@Override public boolean changeToMaster(int masterEpoch) {
final int lastEpoch = this.epochCache.lastEpoch();
if (masterEpoch <= lastEpoch) {
if (masterEpoch < lastEpoch) {
return false;
}
destroyConnections();
@@ -172,6 +168,9 @@ public class AutoSwitchHAService extends DefaultHAService {
final AutoSwitchHAConnection connection = (AutoSwitchHAConnection) haConnection;
final String slaveAddress = connection.getSlaveAddress();
if (currentSyncStateSet.contains(slaveAddress)) {
if (this.defaultMessageStore.getMaxPhyOffset() == connection.getSlaveAckOffset()) {
continue;
}
if ((System.currentTimeMillis() - connection.getLastCatchUpTimeMs()) > haMaxTimeSlaveNotCatchup) {
newSyncStateSet.remove(slaveAddress);
}
test/src/test/java/org/apache/rocketmq/test/autoswitchrole/AutoSwitchRoleBase.java
+1 -1
View File
@@ -123,7 +123,7 @@ public class AutoSwitchRoleBase {
protected ControllerConfig buildControllerConfig(final String id, final String peers) {
final ControllerConfig config = new ControllerConfig();
config.setStartupController(true);
config.setEnableStartupController(true);
config.setControllerDLegerGroup("group1");
config.setControllerDLegerPeers(peers);
config.setControllerDLegerSelfId(id);