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

[ISSUE #7219] Fix Concurrent modify syncStateSet and Mark synchronizing frequently when shrink. (#7220)

Browse Source
This commit is contained in:
echooymxq
2023-08-23 11:11:42 +08:00
committed by GitHub
parent 3884f59594
commit 017ad11047
2 changed files with 28 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/controller/ReplicasManager.java
+16 -13
View File
@@ -542,7 +542,7 @@ public class ReplicasManager {
this.brokerMetadata.updateAndPersist(brokerConfig.getBrokerClusterName(), brokerConfig.getBrokerName(), tempBrokerMetadata.getBrokerId());
this.tempBrokerMetadata.clear();
this.brokerControllerId = this.brokerMetadata.getBrokerId();
this.haService.setBrokerControllerId(this.brokerControllerId);
this.haService.setLocalBrokerId(this.brokerControllerId);
return true;
} catch (Exception e) {
LOGGER.error("fail to create metadata file", e);
@@ -594,7 +594,7 @@ public class ReplicasManager {
if (this.brokerMetadata.isLoaded()) {
this.registerState = RegisterState.CREATE_METADATA_FILE_DONE;
this.brokerControllerId = brokerMetadata.getBrokerId();
this.haService.setBrokerControllerId(this.brokerControllerId);
this.haService.setLocalBrokerId(this.brokerControllerId);
return;
}
// 2. check if temp metadata exist
@@ -735,23 +735,26 @@ public class ReplicasManager {
if (this.checkSyncStateSetTaskFuture != null) {
this.checkSyncStateSetTaskFuture.cancel(false);
}
this.checkSyncStateSetTaskFuture = this.scheduledService.scheduleAtFixedRate(() -> {
checkSyncStateSetAndDoReport();
}, 3 * 1000, this.brokerConfig.getCheckSyncStateSetPeriod(), TimeUnit.MILLISECONDS);
this.checkSyncStateSetTaskFuture = this.scheduledService.scheduleAtFixedRate(this::checkSyncStateSetAndDoReport, 3 * 1000,
this.brokerConfig.getCheckSyncStateSetPeriod(), TimeUnit.MILLISECONDS);
}
private void checkSyncStateSetAndDoReport() {
final Set<Long> newSyncStateSet = this.haService.maybeShrinkSyncStateSet();
newSyncStateSet.add(this.brokerControllerId);
synchronized (this) {
if (this.syncStateSet != null) {
// Check if syncStateSet changed
if (this.syncStateSet.size() == newSyncStateSet.size() && this.syncStateSet.containsAll(newSyncStateSet)) {
return;
try {
final Set<Long> newSyncStateSet = this.haService.maybeShrinkSyncStateSet();
newSyncStateSet.add(this.brokerControllerId);
synchronized (this) {
if (this.syncStateSet != null) {
// Check if syncStateSet changed
if (this.syncStateSet.size() == newSyncStateSet.size() && this.syncStateSet.containsAll(newSyncStateSet)) {
return;
}
}
}
doReportSyncStateSetChanged(newSyncStateSet);
} catch (Exception e) {
LOGGER.error("Check syncStateSet error", e);
}
doReportSyncStateSetChanged(newSyncStateSet);
}
private void doReportSyncStateSetChanged(Set<Long> newSyncStateSet) {
store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAService.java
+12 -9
View File
@@ -41,6 +41,7 @@ import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Iterator;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
@@ -73,7 +74,7 @@ public class AutoSwitchHAService extends DefaultHAService {
private EpochFileCache epochCache;
private AutoSwitchHAClient haClient;
private Long brokerControllerId = null;
private Long localBrokerId = null;
public AutoSwitchHAService() {
}
@@ -287,9 +288,11 @@ public class AutoSwitchHAService extends DefaultHAService {
// If the slaveBrokerId is in syncStateSet but not in connectionCaughtUpTimeTable,
// it means that the broker has not connected.
for (Long slaveBrokerId : newSyncStateSet) {
if (!this.connectionCaughtUpTimeTable.containsKey(slaveBrokerId)) {
newSyncStateSet.remove(slaveBrokerId);
Iterator<Long> iterator = newSyncStateSet.iterator();
while (iterator.hasNext()) {
Long slaveBrokerId = iterator.next();
if (!Objects.equals(slaveBrokerId, this.localBrokerId) && !this.connectionCaughtUpTimeTable.containsKey(slaveBrokerId)) {
iterator.remove();
isSyncStateSetChanged = true;
}
}
@@ -419,7 +422,7 @@ public class AutoSwitchHAService extends DefaultHAService {
// To avoid the syncStateSet is not consistent with connectionList.
// Fix issue: https://github.com/apache/rocketmq/issues/6662
for (Long syncId : currentSyncStateSet) {
if (!idList.contains(syncId) && this.brokerControllerId != null && !Objects.equals(syncId, this.brokerControllerId)) {
if (!idList.contains(syncId) && this.localBrokerId != null && !Objects.equals(syncId, this.localBrokerId)) {
LOGGER.warn("Slave {} is still in syncStateSet, but has lost its connection. So new offset can't be compute.", syncId);
// Without check and re-compute, return the confirmOffset's value directly.
return this.defaultMessageStore.getConfirmOffsetDirectly();
@@ -545,12 +548,12 @@ public class AutoSwitchHAService extends DefaultHAService {
return this.epochCache.getAllEntries();
}
public Long getBrokerControllerId() {
return brokerControllerId;
public Long getLocalBrokerId() {
return localBrokerId;
}
public void setBrokerControllerId(Long brokerControllerId) {
this.brokerControllerId = brokerControllerId;
public void setLocalBrokerId(Long localBrokerId) {
this.localBrokerId = localBrokerId;
}
class AutoSwitchAcceptSocketService extends AcceptSocketService {