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

[ISSUE #6306] Fix unexpected state from slave (#6307)

Browse Source 
* typo int readme[ecosystem]

* fix unexpected state from slave
This commit is contained in:
fujian-zfj
2023-03-10 13:39:28 +08:00
committed by GitHub
parent e599187047
commit 7b23042eb2
2 changed files with 13 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAClient.java
+12 -12
View File
@@ -332,21 +332,21 @@ public class AutoSwitchHAClient extends ServiceThread implements HAClient {
}
}
private boolean reportSlaveOffset(final long offsetToReport) throws IOException {
private boolean reportSlaveOffset(HAConnectionState currentState, final long offsetToReport) throws IOException {
this.transferHeaderBuffer.position(0);
this.transferHeaderBuffer.limit(TRANSFER_HEADER_SIZE);
this.transferHeaderBuffer.putInt(this.currentState.ordinal());
this.transferHeaderBuffer.putInt(currentState.ordinal());
this.transferHeaderBuffer.putLong(offsetToReport);
this.transferHeaderBuffer.flip();
return this.haWriter.write(this.socketChannel, this.transferHeaderBuffer);
}
private boolean reportSlaveMaxOffset() throws IOException {
private boolean reportSlaveMaxOffset(HAConnectionState currentState) throws IOException {
boolean result = true;
final long maxPhyOffset = this.messageStore.getMaxPhyOffset();
if (maxPhyOffset > this.currentReportedOffset) {
this.currentReportedOffset = maxPhyOffset;
result = reportSlaveOffset(this.currentReportedOffset);
result = reportSlaveOffset(currentState, this.currentReportedOffset);
}
return result;
}
@@ -369,11 +369,11 @@ public class AutoSwitchHAClient extends ServiceThread implements HAClient {
return this.socketChannel != null;
}
private boolean transferFromMaster() throws IOException {
private boolean transferFromMaster(HAConnectionState currentState) throws IOException {
boolean result;
if (isTimeToReportOffset()) {
LOGGER.info("Slave report current offset {}", this.currentReportedOffset);
result = reportSlaveOffset(this.currentReportedOffset);
result = reportSlaveOffset(currentState, this.currentReportedOffset);
if (!result) {
return false;
}
@@ -386,7 +386,7 @@ public class AutoSwitchHAClient extends ServiceThread implements HAClient {
return false;
}
return this.reportSlaveMaxOffset();
return this.reportSlaveMaxOffset(currentState);
}
@Override
@@ -415,7 +415,7 @@ public class AutoSwitchHAClient extends ServiceThread implements HAClient {
handshakeWithMaster();
continue;
case TRANSFER:
if (!transferFromMaster()) {
if (!transferFromMaster(HAConnectionState.TRANSFER)) {
closeMasterAndWait();
continue;
}
@@ -445,7 +445,7 @@ public class AutoSwitchHAClient extends ServiceThread implements HAClient {
/**
* Compare the master and slave's epoch file, find consistent point, do truncate.
*/
private boolean doTruncate(List<EpochEntry> masterEpochEntries, long masterEndOffset) throws IOException {
private boolean doTruncate(List<EpochEntry> masterEpochEntries, long masterEndOffset, HAConnectionState currentState) throws IOException {
if (this.epochCache.getEntrySize() == 0) {
// If epochMap is empty, means the broker is a new replicas
LOGGER.info("Slave local epochCache is empty, skip truncate log");
@@ -475,7 +475,7 @@ public class AutoSwitchHAClient extends ServiceThread implements HAClient {
changeCurrentState(HAConnectionState.TRANSFER);
this.currentReportedOffset = truncateOffset;
}
if (!reportSlaveMaxOffset()) {
if (!reportSlaveMaxOffset(currentState)) {
LOGGER.error("AutoSwitchHAClient report max offset to master failed");
return false;
}
@@ -534,7 +534,7 @@ public class AutoSwitchHAClient extends ServiceThread implements HAClient {
byteBufferRead.position(readSocketPos);
AutoSwitchHAClient.this.processPosition += bodySize;
LOGGER.info("Receive handshake, masterMaxPosition {}, masterEpochEntries:{}, try truncate log", masterOffset, epochEntries);
if (!doTruncate(epochEntries, masterOffset)) {
if (!doTruncate(epochEntries, masterOffset, HAConnectionState.HANDSHAKE)) {
waitForRunning(1000 * 2);
LOGGER.error("AutoSwitchHAClient truncate log failed in handshake state");
return false;
@@ -573,7 +573,7 @@ public class AutoSwitchHAClient extends ServiceThread implements HAClient {
haService.updateConfirmOffset(Math.min(confirmOffset, messageStore.getMaxPhyOffset()));
if (!reportSlaveMaxOffset()) {
if (!reportSlaveMaxOffset(HAConnectionState.TRANSFER)) {
LOGGER.error("AutoSwitchHAClient report max offset to master failed");
return false;
}
store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAConnection.java
+1 -1
View File
@@ -363,7 +363,7 @@ public class AutoSwitchHAConnection implements HAConnection {
break;
default:
LOGGER.error("Current state illegal {}", currentState);
break;
return false;
}
if (!slaveState.equals(currentState)) {