Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Polish switching logic and auto switch ha code #4406

Merged
merged 3 commits into from
Jun 4, 2022
Merged

Polish switching logic and auto switch ha code #4406

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4eb9f64
Polish switching logic and auto switch ha code
RongtongJin Jun 3, 2022
43ef160
Make UT can pass
RongtongJin Jun 3, 2022
50cd646
Polish the code
RongtongJin Jun 3, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 7 additions & 4 deletions broker/src/main/java/org/apache/rocketmq/broker/hacontroller/ReplicasManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -169,6 +169,9 @@ public void changeToMaster(final int newMasterEpoch, final int syncStateSetEpoch
// Handle the slave synchronise
handleSlaveSynchronize(BrokerRole.SYNC_MASTER);

// Notify ha service, change to master
this.haService.changeToMaster(newMasterEpoch);

this.executorService.submit(() -> {
// Register broker to name-srv
try {
Expand All @@ -177,8 +180,6 @@ public void changeToMaster(final int newMasterEpoch, final int syncStateSetEpoch
LOGGER.error("Error happen when register broker to name-srv, Failed to change broker to master", e);
return;
}
// Notify ha service, change to master
this.haService.changeToMaster(newMasterEpoch);
LOGGER.info("Change broker {} to master success, masterEpoch {}, syncStateSetEpoch:{}", this.localAddress, newMasterEpoch, syncStateSetEpoch);
});
}
Expand All @@ -193,6 +194,7 @@ public void changeToSlave(final String newMasterAddress, final int newMasterEpoc
// Change record
this.masterAddress = newMasterAddress;
this.masterEpoch = newMasterEpoch;

stopCheckSyncStateSet();

// Change config
Expand All @@ -203,6 +205,9 @@ public void changeToSlave(final String newMasterAddress, final int newMasterEpoc
// Handle the slave synchronise
handleSlaveSynchronize(BrokerRole.SLAVE);

// Notify ha service, change to slave
this.haService.changeToSlave(newMasterAddress, newMasterEpoch, this.brokerConfig.getBrokerId());

this.executorService.submit(() -> {
// Register broker to name-srv
try {
Expand All @@ -212,8 +217,6 @@ public void changeToSlave(final String newMasterAddress, final int newMasterEpoc
return;
}

// Notify ha service, change to slave
this.haService.changeToSlave(newMasterAddress, newMasterEpoch, this.brokerConfig.getBrokerId());
LOGGER.info("Change broker {} to slave, newMasterAddress:{}, newMasterEpoch:{}", this.localAddress, newMasterAddress, newMasterEpoch);
});
}
Expand Down
2 changes: 2 additions & 0 deletions store/src/main/java/org/apache/rocketmq/store/CommitLog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -352,6 +352,8 @@ else if (!dispatchRequest.isSuccess()) {
log.warn("maxPhyOffsetOfConsumeQueue({}) >= processOffset({}), truncate dirty logic files", maxPhyOffsetOfConsumeQueue, processOffset);
this.defaultMessageStore.truncateDirtyLogicFiles(processOffset);
}


} else {
// Commitlog case files are deleted
log.warn("The commitlog files are deleted, and delete the consume queue files");
Expand Down
4 changes: 3 additions & 1 deletion store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAConnection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -328,7 +328,9 @@ protected boolean processReadResult(ByteBuffer byteBufferRead) {
slaveRequestOffset = slaveMaxOffset;
}
byteBufferRead.position(readSocketPos);
maybeExpandInSyncStateSet(slaveMaxOffset);
if (!haService.getSyncStateSet().contains(slaveAddress)) {
maybeExpandInSyncStateSet(slaveMaxOffset);
}
AutoSwitchHAConnection.this.haService.notifyTransferSome(AutoSwitchHAConnection.this.slaveAckOffset);
LOGGER.info("slave[" + clientAddress + "] request offset " + slaveMaxOffset);
break;
Expand Down
4 changes: 1 addition & 3 deletions store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,9 +100,6 @@ public AutoSwitchHAService() {

this.defaultMessageStore.recoverTopicQueueTable();

final HashSet<String> newSyncStateSet = new HashSet<>();
newSyncStateSet.add(this.localAddress);
setSyncStateSet(newSyncStateSet);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

在 controller 里的逻辑是, 新选举出来的 master, syncStateSet 只有该 Master, 这里是为了应和 controller 里的逻辑.

LOGGER.info("Change ha to master success, newMasterEpoch:{}, startOffset:{}", masterEpoch, newEpochEntry.getStartOffset());
return true;
}
Expand All @@ -122,6 +119,7 @@ public AutoSwitchHAService() {
this.haClient.setLocalAddress(this.localAddress);
this.haClient.updateSlaveId(slaveId);
this.haClient.updateMasterAddress(newMasterAddr);
this.haClient.updateHaMasterAddress(null);
this.haClient.start();
LOGGER.info("Change ha to slave success, newMasterAddress:{}, newMasterEpoch:{}", newMasterAddr, newMasterEpoch);
return true;
Expand Down