This is an automated email from the ASF dual-hosted git repository.
RongtongJin pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new 78b96bc5e2 [ISSUE #10853] fix(store): validate HA state ordinals
(#11200)
78b96bc5e2 is described below
commit 78b96bc5e21216cd7896efae08f90c5cde4cae53
Author: beautyarbutin <[email protected]>
AuthorDate: Wed Sep 23 14:42:31 2026 +0800
[ISSUE #10853] fix(store): validate HA state ordinals (#11200)
* [ISSUE #10853] fix(store): validate HA state ordinals
* test(store): cover valid HA state ordinals
---
.../rocketmq/store/ha/HAConnectionState.java | 8 +-
.../store/ha/autoswitch/AutoSwitchHAClient.java | 13 ++-
.../ha/autoswitch/AutoSwitchHAConnection.java | 7 +-
.../rocketmq/store/ha/HAConnectionStateTest.java} | 37 +++---
.../AutoSwitchHAStateValidationTest.java | 125 +++++++++++++++++++++
5 files changed, 162 insertions(+), 28 deletions(-)
diff --git
a/store/src/main/java/org/apache/rocketmq/store/ha/HAConnectionState.java
b/store/src/main/java/org/apache/rocketmq/store/ha/HAConnectionState.java
index 4f0c5ca909..7a3f2eb903 100644
--- a/store/src/main/java/org/apache/rocketmq/store/ha/HAConnectionState.java
+++ b/store/src/main/java/org/apache/rocketmq/store/ha/HAConnectionState.java
@@ -37,5 +37,11 @@ public enum HAConnectionState {
/**
* Connection shutdown.
*/
- SHUTDOWN,
+ SHUTDOWN;
+
+ private static final HAConnectionState[] VALUES = values();
+
+ public static HAConnectionState fromOrdinal(int ordinal) {
+ return ordinal >= 0 && ordinal < VALUES.length ? VALUES[ordinal] :
null;
+ }
}
diff --git
a/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAClient.java
b/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAClient.java
index 3dd14f4e35..665d9aae3b 100644
---
a/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAClient.java
+++
b/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAClient.java
@@ -494,17 +494,22 @@ public class AutoSwitchHAClient extends ServiceThread
implements HAClient {
int masterEpoch =
byteBufferRead.getInt(processPosition +
AutoSwitchHAConnection.HANDSHAKE_HEADER_SIZE - 4);
long masterEpochStartOffset = 0;
long confirmOffset = 0;
+ HAConnectionState masterConnectionState =
HAConnectionState.fromOrdinal(masterState);
+ if (masterConnectionState == null) {
+ LOGGER.error("Received illegal master state
ordinal {}", masterState);
+ return false;
+ }
// If master send transfer header data, set
masterEpochStartOffset and confirmOffset value.
- if (masterState ==
HAConnectionState.TRANSFER.ordinal() && diff >=
AutoSwitchHAConnection.TRANSFER_HEADER_SIZE) {
+ if (masterConnectionState ==
HAConnectionState.TRANSFER && diff >=
AutoSwitchHAConnection.TRANSFER_HEADER_SIZE) {
masterEpochStartOffset =
byteBufferRead.getLong(processPosition +
AutoSwitchHAConnection.TRANSFER_HEADER_SIZE - 16);
confirmOffset =
byteBufferRead.getLong(processPosition +
AutoSwitchHAConnection.TRANSFER_HEADER_SIZE - 8);
}
- if (masterState !=
AutoSwitchHAClient.this.currentState.ordinal()) {
- int headerSize = masterState ==
HAConnectionState.TRANSFER.ordinal() ?
AutoSwitchHAConnection.TRANSFER_HEADER_SIZE :
AutoSwitchHAConnection.HANDSHAKE_HEADER_SIZE;
+ if (masterConnectionState !=
AutoSwitchHAClient.this.currentState) {
+ int headerSize = masterConnectionState ==
HAConnectionState.TRANSFER ? AutoSwitchHAConnection.TRANSFER_HEADER_SIZE :
AutoSwitchHAConnection.HANDSHAKE_HEADER_SIZE;
AutoSwitchHAClient.this.processPosition +=
headerSize + bodySize;
AutoSwitchHAClient.this.waitForRunning(1);
LOGGER.error("State not matched, masterState:{},
slaveState:{}, bodySize:{}, offset:{}, masterEpoch:{},
masterEpochStartOffset:{}, confirmOffset:{}",
- HAConnectionState.values()[masterState],
AutoSwitchHAClient.this.currentState, bodySize, masterOffset, masterEpoch,
masterEpochStartOffset, confirmOffset);
+ masterConnectionState,
AutoSwitchHAClient.this.currentState, bodySize, masterOffset, masterEpoch,
masterEpochStartOffset, confirmOffset);
return false;
}
diff --git
a/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAConnection.java
b/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAConnection.java
index cc55937aeb..e67ce46b0c 100644
---
a/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAConnection.java
+++
b/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAConnection.java
@@ -310,7 +310,12 @@ public class AutoSwitchHAConnection implements
HAConnection {
int diff = byteBufferRead.position() -
ReadSocketService.this.processPosition;
if (diff >= AutoSwitchHAClient.MIN_HEADER_SIZE) {
int readPosition =
ReadSocketService.this.processPosition;
- HAConnectionState slaveState =
HAConnectionState.values()[byteBufferRead.getInt(readPosition)];
+ int slaveStateOrdinal =
byteBufferRead.getInt(readPosition);
+ HAConnectionState slaveState =
HAConnectionState.fromOrdinal(slaveStateOrdinal);
+ if (slaveState == null) {
+ LOGGER.error("Received illegal slave state ordinal
{}", slaveStateOrdinal);
+ return false;
+ }
switch (slaveState) {
case HANDSHAKE:
diff --git
a/store/src/main/java/org/apache/rocketmq/store/ha/HAConnectionState.java
b/store/src/test/java/org/apache/rocketmq/store/ha/HAConnectionStateTest.java
similarity index 62%
copy from
store/src/main/java/org/apache/rocketmq/store/ha/HAConnectionState.java
copy to
store/src/test/java/org/apache/rocketmq/store/ha/HAConnectionStateTest.java
index 4f0c5ca909..7b32c7a26a 100644
--- a/store/src/main/java/org/apache/rocketmq/store/ha/HAConnectionState.java
+++
b/store/src/test/java/org/apache/rocketmq/store/ha/HAConnectionStateTest.java
@@ -14,28 +14,21 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.rocketmq.store.ha;
-public enum HAConnectionState {
- /**
- * Ready to start connection.
- */
- READY,
- /**
- * CommitLog consistency checking.
- */
- HANDSHAKE,
- /**
- * Synchronizing data.
- */
- TRANSFER,
- /**
- * Temporarily stop transferring.
- */
- SUSPEND,
- /**
- * Connection shutdown.
- */
- SHUTDOWN,
+import org.junit.Test;
+
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+
+public class HAConnectionStateTest {
+
+ @Test
+ public void testFromOrdinal() {
+ for (HAConnectionState state : HAConnectionState.values()) {
+ assertSame(state, HAConnectionState.fromOrdinal(state.ordinal()));
+ }
+ assertNull(HAConnectionState.fromOrdinal(-1));
+
assertNull(HAConnectionState.fromOrdinal(HAConnectionState.values().length));
+ }
}
diff --git
a/store/src/test/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAStateValidationTest.java
b/store/src/test/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAStateValidationTest.java
new file mode 100644
index 0000000000..c7ed78058d
--- /dev/null
+++
b/store/src/test/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAStateValidationTest.java
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.store.ha.autoswitch;
+
+import java.lang.reflect.Field;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.rocketmq.common.BrokerConfig;
+import org.apache.rocketmq.store.DefaultMessageStore;
+import org.apache.rocketmq.store.config.MessageStoreConfig;
+import org.apache.rocketmq.store.ha.HAConnectionState;
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class AutoSwitchHAStateValidationTest {
+
+ @Test
+ public void testServerReaderRejectsUnknownSlaveStateOrdinal() throws
Exception {
+ assertFalse(processSlaveState(Integer.MAX_VALUE));
+ }
+
+ @Test
+ public void testServerReaderAcceptsKnownSlaveStateOrdinal() throws
Exception {
+ assertTrue(processSlaveState(HAConnectionState.HANDSHAKE.ordinal()));
+ }
+
+ private boolean processSlaveState(int stateOrdinal) throws Exception {
+ AutoSwitchHAService haService = mock(AutoSwitchHAService.class);
+ DefaultMessageStore messageStore = mock(DefaultMessageStore.class);
+ when(haService.getConnectionCount()).thenReturn(new AtomicInteger());
+ when(haService.getDefaultMessageStore()).thenReturn(messageStore);
+ when(messageStore.getBrokerConfig()).thenReturn(new BrokerConfig());
+ when(messageStore.getMessageStoreConfig()).thenReturn(new
MessageStoreConfig());
+
+ try (ServerSocketChannel serverSocket = ServerSocketChannel.open();
+ SocketChannel slaveSocket = SocketChannel.open()) {
+ serverSocket.bind(new InetSocketAddress("127.0.0.1", 0));
+ slaveSocket.connect(serverSocket.getLocalAddress());
+
+ try (SocketChannel masterSocket = serverSocket.accept()) {
+ AutoSwitchHAConnection connection = new AutoSwitchHAConnection(
+ haService, masterSocket, mock(EpochFileCache.class));
+ try {
+ AutoSwitchHAConnection.ReadSocketService readSocketService
= getReadSocketService(connection);
+ AutoSwitchHAConnection.ReadSocketService.HAServerReader
reader = readSocketService.new HAServerReader();
+ ByteBuffer frame =
ByteBuffer.allocate(AutoSwitchHAClient.HANDSHAKE_HEADER_SIZE);
+ frame.putInt(stateOrdinal);
+ frame.putShort((short) 0);
+ frame.putShort((short) 0);
+ frame.putLong(1L);
+
+ return reader.processReadResult(frame);
+ } finally {
+ connection.shutdown();
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testClientReaderRejectsUnknownMasterStateOrdinal() throws
Exception {
+ assertFalse(processMasterState(Integer.MAX_VALUE,
AutoSwitchHAConnection.HANDSHAKE_HEADER_SIZE));
+ }
+
+ @Test
+ public void testClientReaderHandlesKnownMasterStateOrdinals() throws
Exception {
+ assertFalse(processMasterState(
+ HAConnectionState.HANDSHAKE.ordinal(),
AutoSwitchHAConnection.HANDSHAKE_HEADER_SIZE));
+ assertFalse(processMasterState(
+ HAConnectionState.TRANSFER.ordinal(),
AutoSwitchHAConnection.TRANSFER_HEADER_SIZE));
+ }
+
+ private boolean processMasterState(int stateOrdinal, int headerSize)
throws Exception {
+ AutoSwitchHAService haService = mock(AutoSwitchHAService.class);
+ DefaultMessageStore messageStore = mock(DefaultMessageStore.class);
+ when(haService.getDefaultMessageStore()).thenReturn(messageStore);
+ when(messageStore.getBrokerConfig()).thenReturn(new BrokerConfig());
+ when(messageStore.getMessageStoreConfig()).thenReturn(new
MessageStoreConfig());
+ AutoSwitchHAClient client = new AutoSwitchHAClient(
+ haService, messageStore, mock(EpochFileCache.class), 1L);
+ try {
+ ByteBuffer frame = ByteBuffer.allocate(headerSize);
+ frame.putInt(stateOrdinal);
+ frame.putInt(0);
+ frame.putLong(0);
+ frame.putInt(0);
+ if (headerSize == AutoSwitchHAConnection.TRANSFER_HEADER_SIZE) {
+ frame.putLong(0);
+ frame.putLong(0);
+ }
+
+ return client.new HAClientReader().processReadResult(frame);
+ } finally {
+ client.shutdown();
+ }
+ }
+
+ private AutoSwitchHAConnection.ReadSocketService getReadSocketService(
+ AutoSwitchHAConnection connection) throws ReflectiveOperationException
{
+ Field field =
AutoSwitchHAConnection.class.getDeclaredField("readSocketService");
+ field.setAccessible(true);
+ return (AutoSwitchHAConnection.ReadSocketService)
field.get(connection);
+ }
+}
