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

Fix flaky HATest semi-sync replication

Browse Source
This commit is contained in:
RongtongJin
2026-06-13 09:56:59 +08:00
parent 86d1df4917
commit 3b41597e2f
1 changed files with 17 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
store/src/test/java/org/apache/rocketmq/store/HATest.java
+17
View File
@@ -25,6 +25,7 @@ import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.time.Duration;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
@@ -38,6 +39,7 @@ import org.apache.rocketmq.common.message.MessageExtBrokerInner;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.config.FlushDiskType;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.ha.HAConnection;
import org.apache.rocketmq.store.ha.HAConnectionState;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
import org.junit.After;
@@ -104,6 +106,7 @@ public class HATest {
slaveMessageStore.start();
slaveMessageStore.updateHaMasterAddress("127.0.0.1:" + masterMessageStoreConfig.getHaListenPort());
await().atMost(6, SECONDS).until(() -> slaveMessageStore.getHaService().getHAClient().getCurrentState() == HAConnectionState.TRANSFER);
await().atMost(6, SECONDS).until(this::isSlaveReadyForReplication);
}
@Test
@@ -281,6 +284,20 @@ public class HATest {
return msg;
}
private boolean isSlaveReadyForReplication() {
if (slaveMessageStore.getHaService().getHAClient().getCurrentState() != HAConnectionState.TRANSFER) {
return false;
}
long slaveMaxOffset = slaveMessageStore.getMaxPhyOffset();
List<HAConnection> connections = messageStore.getHaService().getConnectionList();
synchronized (connections) {
return connections.stream().anyMatch(connection ->
connection.getCurrentState() == HAConnectionState.TRANSFER
&& connection.getSlaveAckOffset() >= slaveMaxOffset);
}
}
private boolean isCommitLogAvailable(DefaultMessageStore store) {
try {
Field serviceField = store.getClass().getDeclaredField("reputMessageService");