Repository: AutoMQ/automq (language: java)

Issue:
[Enhancement] Support automq custom records parsing in metadata shell
### Who is this for and what problem do they have today?

For developers and users working with the KRaft system that involves custom metadata, a problem they face is the difficulty in parsing this custom metadata within KRaft, which complicates the troubleshooting process.

### Why is solving this problem impactful?

It improves the debugging and troubleshooting process for developers.

### Additional notes

Example of file hierarchy:
```
.
├── kvRecords
│   └── _kafka_FVWxwMClSYObxlUORZANzA
│       └── 4IHvQLhZSJ6szHPcyady3Q
│           └── 0
├── nodes
│   └── 1 // node id
│       ├── s3StreamSetObjects
│       │   └── 4 // object id
│       │       ├── dataTimeInMs
│       │       ├── orderId
│       │       └── ranges
│       └── walMetadata
│           ├── failoverMode
│           └── nodeEpoch
├── s3Objects
│   └── 1 // object id
│       ├── committedTimeInMs
│       ├── expiredTimeInMs
│       ├── markDestroyedTimeInMs
│       ├── objectSize
│       ├── objectState
│       └── preparedTimeInMs
└── s3Streams
    └── 0 // stream id
        ├── epoch
        ├── rangeIndex
        ├── ranges
        │   └── 0 // range index
        │       ├── endOffset
        │       ├── epoch
        │       ├── nodeId
        │       └── startOffset
        ├── s3StreamObjects
        │   └── 3 // object id
        │       ├── dataTimeInMs
        │       ├── endOffset
        │       └── startOffset
        ├── startOffset
        └── streamState
```

BTW, in most case, the key of `kvRecords` contains '/', so we have to create a directory rather than a single file to represent the key...


Discussion / hints:
s3StreamSetObjects.ranges may convert to human readable string when cat

Fix patch:
diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index 61fe82f0fa..e42dcad707 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -35,6 +35,7 @@ jobs:
               - '.github/workflows/**'
               - 'core/**'
               - 'metadata/**'
+              - 'shell/**'
   build-kos:
     needs: [ paths-filter ]
     if: ${{ needs.paths-filter.outputs.build-kos == 'true' || github.event_name == 'push' }}
diff --git a/build.gradle b/build.gradle
index 1c7629805f..5af206dc89 100644
--- a/build.gradle
+++ b/build.gradle
@@ -1990,6 +1990,12 @@ project(':shell') {
     implementation libs.jacksonJDK8Datatypes
     implementation libs.jline
     implementation libs.slf4jApi
+    implementation(extLibs.s3stream) {
+      exclude group: 'io.netty', module: '*'
+      exclude group: 'org.slf4j', module: '*'
+      exclude group: 'net.sourceforge.argparse4j', module: '*'
+      version branchVersions.s3stream
+    }
     implementation project(':server-common')
     implementation project(':clients')
     implementation project(':core')
diff --git a/shell/src/main/java/org/apache/kafka/shell/MetadataNodeManager.java b/shell/src/main/java/org/apache/kafka/shell/MetadataNodeManager.java
index 93dc3073fd..2036fbbde0 100644
--- a/shell/src/main/java/org/apache/kafka/shell/MetadataNodeManager.java
+++ b/shell/src/main/java/org/apache/kafka/shell/MetadataNodeManager.java
@@ -17,12 +17,17 @@
 
 package org.apache.kafka.shell;
 
+import com.automq.stream.s3.metadata.StreamOffsetRange;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
+import java.util.Arrays;
+import java.util.stream.Stream;
 import org.apache.kafka.common.config.ConfigResource;
 import org.apache.kafka.common.metadata.AccessControlEntryRecord;
 import org.apache.kafka.common.metadata.AccessControlEntryRecordJsonConverter;
+import org.apache.kafka.common.metadata.AssignedS3ObjectIdRecord;
+import org.apache.kafka.common.metadata.AssignedStreamIdRecord;
 import org.apache.kafka.common.metadata.BrokerRegistrationChangeRecord;
 import org.apache.kafka.common.metadata.ClientQuotaRecord;
 import org.apache.kafka.common.metadata.ClientQuotaRecord.EntityData;
@@ -30,23 +35,44 @@
 import org.apache.kafka.common.metadata.FeatureLevelRecord;
 import org.apache.kafka.common.metadata.FeatureLevelRecordJsonConverter;
 import org.apache.kafka.common.metadata.FenceBrokerRecord;
+import org.apache.kafka.common.metadata.KVRecord;
 import org.apache.kafka.common.metadata.MetadataRecordType;
+import org.apache.kafka.common.metadata.NodeWALMetadataRecord;
+import org.apache.kafka.common.metadata.NodeWALMetadataRecordJsonConverter;
 import org.apache.kafka.common.metadata.PartitionChangeRecord;
 import org.apache.kafka.common.metadata.PartitionRecord;
 import org.apache.kafka.common.metadata.PartitionRecordJsonConverter;
 import org.apache.kafka.common.metadata.ProducerIdsRecord;
+import org.apache.kafka.common.metadata.RangeRecord;
+import org.apache.kafka.common.metadata.RangeRecordJsonConverter;
 import org.apache.kafka.common.metadata.RegisterBrokerRecord;
 import org.apache.kafka.common.metadata.RemoveAccessControlEntryRecord;
+import org.apache.kafka.common.metadata.RemoveKVRecord;
+import org.apache.kafka.common.metadata.RemoveNodeWALMetadataRecord;
+import org.apache.kafka.common.metadata.RemoveRangeRecord;
+import org.apache.kafka.common.metadata.RemoveS3ObjectRecord;
+import org.apache.kafka.common.metadata.RemoveS3StreamObjectRecord;
+import org.apache.kafka.common.metadata.RemoveS3StreamRecord;
+import org.apache.kafka.common.metadata.RemoveStreamSetObjectRecord;
 import org.apache.kafka.common.metadata.RemoveTopicRecord;
+import org.apache.kafka.common.metadata.S3ObjectRecord;
+import org.apache.kafka.common.metadata.S3ObjectRecordJsonConverter;
+import org.apache.kafka.common.metadata.S3StreamObjectRecord;
+import org.apache.kafka.common.metadata.S3StreamObjectRecordJsonConverter;
+import org.apache.kafka.common.metadata.S3StreamRecord;
+import org.apache.kafka.common.metadata.S3StreamRecordJsonConverter;
+import org.apache.kafka.common.metadata.S3StreamSetObjectRecord;
 import org.apache.kafka.common.metadata.TopicRecord;
 import org.apache.kafka.common.metadata.UnfenceBrokerRecord;
 import org.apache.kafka.common.metadata.UnregisterBrokerRecord;
+import org.apache.kafka.common.metadata.UpdateNextNodeIdRecord;
 import org.apache.kafka.common.protocol.ApiMessage;
 import org.apache.kafka.common.utils.AppInfoParser;
 import org.apache.kafka.common.utils.LogContext;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.metadata.BrokerRegistrationFencingChange;
 import org.apache.kafka.metadata.BrokerRegistrationInControlledShutdownChange;
+import org.apache.kafka.metadata.stream.S3StreamSetObject;
 import org.apache.kafka.queue.EventQueue;
 import org.apache.kafka.queue.KafkaEventQueue;
 import org.apache.kafka.raft.Batch;
@@ -387,7 +413,7 @@ private void handleCommitImpl(MetadataRecordType type, ApiMessage message)
                 break;
             }
             default:
-                throw new RuntimeException("Unhandled metadata record type");
+                handleExtCommitImpl(type, message);
         }
     }
 
@@ -403,4 +429,162 @@ static List<String> clientQuotaRecordDirectories(List<EntityData> entityData) {
         }
         return result;
     }
+
+    /**
+     * <h3>File Hierarchy for AutoMQ Custom Metadata</h3>
+     * <pre>
+     * .
+     * ├── kvRecords
+     * │   └── _kafka_FVWxwMClSYObxlUORZANzA
+     * │       └── 4IHvQLhZSJ6szHPcyady3Q
+     * │           └── 0
+     * ├── nodes
+     * │   ├── 1 - node id
+     * │   │   ├── s3StreamSetObjects
+     * │   │   │   └── 4 - object id
+     * │   │   │       ├── dataTimeInMs
+     * │   │   │       ├── orderId
+     * │   │   │       └── ranges
+     * │   │   └── walMetadata
+     * │   └── nextNodeId
+     * ├── s3Objects
+     * │   ├── 1 - object id
+     * │   └── nextObjectId
+     * └── s3Streams
+     *     ├── 0 - stream id
+     *     │   ├── ranges
+     *     │   │   └── 0
+     *     │   └── s3StreamObjects
+     *     │       └── 3
+     *     └── nextStreamId
+     * </pre>
+     */
+    private void handleExtCommitImpl(MetadataRecordType type, ApiMessage message) {
+        switch (type) {
+            // For AutoMQ-related records.
+            case S3_STREAM_RECORD: {
+                S3StreamRecord record = (S3StreamRecord) message;
+                DirectoryNode s3Streams = data.root.mkdirs("s3Streams");
+                DirectoryNode s3Stream = s3Streams.mkdirs(Long.toString(record.streamId()));
+                String data = S3StreamRecordJsonConverter.write(record, S3StreamRecord.HIGHEST_SUPPORTED_VERSION).toPrettyString();
+                s3Stream.create("data").setContents(data);
+                break;
+            }
+            case REMOVE_S3_STREAM_RECORD: {
+                RemoveS3StreamRecord record = (RemoveS3StreamRecord) message;
+                data.root.rmrf("s3Streams", Long.toString(record.streamId()));
+                break;
+            }
+            case RANGE_RECORD: {
+                RangeRecord record = (RangeRecord) message;
+                DirectoryNode streams = data.root.mkdirs("s3Streams");
+                DirectoryNode stream = streams.mkdirs(Long.toString(record.streamId()));
+                DirectoryNode ranges = stream.mkdirs("ranges");
+                FileNode file = ranges.create(Integer.toString(record.rangeIndex()));
+                file.setContents(RangeRecordJsonConverter.write(record, RangeRecord.HIGHEST_SUPPORTED_VERSION).toPrettyString());
+                break;
+            }
+            case REMOVE_RANGE_RECORD: {
+                RemoveRangeRecord record = (RemoveRangeRecord) message;
+                data.root.rmrf("s3Streams", Long.toString(record.streamId()), "ranges", Integer.toString(record.rangeIndex()));
+                break;
+            }
+            case S3_STREAM_OBJECT_RECORD: {
+                S3StreamObjectRecord record = (S3StreamObjectRecord) message;
+                DirectoryNode s3StreamObjects = data.root.mkdirs("s3Streams", Long.toString(record.streamId()), "s3StreamObjects");
+                FileNode file = s3StreamObjects.create(Long.toString(record.objectId()));
+                file.setContents(S3StreamObjectRecordJsonConverter.write(record, S3StreamObjectRecord.HIGHEST_SUPPORTED_VERSION).toPrettyString());
+                break;
+            }
+            case REMOVE_S3_STREAM_OBJECT_RECORD: {
+                RemoveS3StreamObjectRecord record = (RemoveS3StreamObjectRecord) message;
+                data.root.rmrf("s3Streams", Long.toString(record.streamId()), "s3StreamObjects", Long.toString(record.objectId()));
+                break;
+            }
+            case S3_STREAM_SET_OBJECT_RECORD: {
+                // TODO: a better solution is to convert it to a JSON string and store it in one file, but ranges will not be human-readable.
+                S3StreamSetObjectRecord record = (S3StreamSetObjectRecord) message;
+                DirectoryNode streamSetObject = data.root.mkdirs("nodes", Integer.toString(record.nodeId()), "s3StreamSetObjects", Long.toString(record.objectId()));
+                streamSetObject.create("orderId").setContents(Long.toString(record.orderId()));
+                streamSetObject.create("dataTimeInMs").setContents(Long.toString(record.dataTimeInMs()));
+                byte[] bytes = record.ranges();
+                // Decode ranges and convert it to human-readable string.
+                List<StreamOffsetRange> ranges = S3StreamSetObject.decode(bytes);
+                streamSetObject.create("ranges").setContents(ranges.toString());
+                break;
+            }
+            case REMOVE_STREAM_SET_OBJECT_RECORD: {
+                RemoveStreamSetObjectRecord record = (RemoveStreamSetObjectRecord) message;
+                data.root.rmrf("nodes", Integer.toString(record.nodeId()), "s3StreamSetObjects", Long.toString(record.objectId()));
+                break;
+            }
+            case S3_OBJECT_RECORD: {
+                S3ObjectRecord record = (S3ObjectRecord) message;
+                DirectoryNode s3Objects = data.root.mkdirs("s3Objects");
+                FileNode file = s3Objects.create(Long.toString(record.objectId()));
+                file.setContents(S3ObjectRecordJsonConverter.write(record, S3ObjectRecord.HIGHEST_SUPPORTED_VERSION).toPrettyString());
+                break;
+            }
+            case REMOVE_S3_OBJECT_RECORD: {
+                RemoveS3ObjectRecord record = (RemoveS3ObjectRecord) message;
+                data.root.rmrf("s3Objects", Long.toString(record.objectId()));
+                break;
+            }
+            case ASSIGNED_STREAM_ID_RECORD: {
+                AssignedStreamIdRecord record = (AssignedStreamIdRecord) message;
+                DirectoryNode streamIdsNode = data.root.mkdirs("s3Streams");
+                streamIdsNode.create("nextStreamId").setContents(Long.toString(record.assignedStreamId()));
+                break;
+            }
+            case ASSIGNED_S3_OBJECT_ID_RECORD: {
+                AssignedS3ObjectIdRecord record = (AssignedS3ObjectIdRecord) message;
+                DirectoryNode s3ObjectsNode = data.root.mkdirs("s3Objects");
+                s3ObjectsNode.create("nextObjectId").setContents(Long.toString(record.assignedS3ObjectId()));
+                break;
+            }
+            case NODE_WALMETADATA_RECORD: {
+                NodeWALMetadataRecord record = (NodeWALMetadataRecord) message;
+                DirectoryNode nodeId = data.root.mkdirs("nodes", Integer.toString(record.nodeId()));
+                FileNode file = nodeId.create("walMetadata");
+                file.setContents(NodeWALMetadataRecordJsonConverter.write(record, NodeWALMetadataRecord.HIGHEST_SUPPORTED_VERSION).toPrettyString());
+                break;
+            }
+            case REMOVE_NODE_WALMETADATA_RECORD: {
+                RemoveNodeWALMetadataRecord record = (RemoveNodeWALMetadataRecord) message;
+                data.root.rmrf("nodes", Integer.toString(record.nodeId()), "walMetadata");
+                break;
+            }
+            case KVRECORD: {
+                KVRecord record = (KVRecord) message;
+                DirectoryNode kvRecords = data.root.mkdirs("kvRecords");
+                for (KVRecord.KeyValue kv : record.keyValues()) {
+                    DirectoryNode node = kvRecords;
+                    String fileName = kv.key();
+                    String[] filePathArray = kv.key().split("/");
+                    if (filePathArray.length > 1) {
+                        String[] dirPathArray = Arrays.copyOfRange(filePathArray, 0, filePathArray.length - 1);
+                        node = node.mkdirs(dirPathArray);
+                        fileName = filePathArray[filePathArray.length - 1];
+                    }
+                    node.create(fileName).setContents(Arrays.toString(kv.value()));
+                }
+                break;
+            }
+            case REMOVE_KVRECORD: {
+                RemoveKVRecord record = (RemoveKVRecord) message;
+                for (String key : record.keys()) {
+                    String[] pathArray = Stream.concat(Stream.of("kvRecords"), Stream.of(key.split("/"))).toArray(String[]::new);
+                    data.root.rmrf(pathArray);
+                }
+                break;
+            }
+            case UPDATE_NEXT_NODE_ID_RECORD: {
+                UpdateNextNodeIdRecord record = (UpdateNextNodeIdRecord) message;
+                DirectoryNode nodes = data.root.mkdirs("nodes");
+                nodes.create("nextNodeId").setContents(Integer.toString(record.nodeId()));
+                break;
+            }
+        }
+
+    }
 }


Test patch:
diff --git a/shell/src/test/java/org/apache/kafka/shell/ExtMetadataNodeManagerTest.java b/shell/src/test/java/org/apache/kafka/shell/ExtMetadataNodeManagerTest.java
new file mode 100644
index 0000000000..6e69fcd0c3
--- /dev/null
+++ b/shell/src/test/java/org/apache/kafka/shell/ExtMetadataNodeManagerTest.java
@@ -0,0 +1,250 @@
+/*
+ * 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.kafka.shell;
+
+import com.automq.stream.s3.metadata.StreamOffsetRange;
+import com.automq.stream.s3.metadata.StreamState;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.kafka.common.metadata.AssignedS3ObjectIdRecord;
+import org.apache.kafka.common.metadata.AssignedStreamIdRecord;
+import org.apache.kafka.common.metadata.KVRecord;
+import org.apache.kafka.common.metadata.NodeWALMetadataRecord;
+import org.apache.kafka.common.metadata.NodeWALMetadataRecordJsonConverter;
+import org.apache.kafka.common.metadata.RangeRecord;
+import org.apache.kafka.common.metadata.RangeRecordJsonConverter;
+import org.apache.kafka.common.metadata.RemoveKVRecord;
+import org.apache.kafka.common.metadata.RemoveNodeWALMetadataRecord;
+import org.apache.kafka.common.metadata.RemoveRangeRecord;
+import org.apache.kafka.common.metadata.RemoveS3ObjectRecord;
+import org.apache.kafka.common.metadata.RemoveS3StreamObjectRecord;
+import org.apache.kafka.common.metadata.RemoveS3StreamRecord;
+import org.apache.kafka.common.metadata.RemoveStreamSetObjectRecord;
+import org.apache.kafka.common.metadata.S3ObjectRecord;
+import org.apache.kafka.common.metadata.S3ObjectRecordJsonConverter;
+import org.apache.kafka.common.metadata.S3StreamObjectRecord;
+import org.apache.kafka.common.metadata.S3StreamObjectRecordJsonConverter;
+import org.apache.kafka.common.metadata.S3StreamRecord;
+import org.apache.kafka.common.metadata.S3StreamRecordJsonConverter;
+import org.apache.kafka.common.metadata.S3StreamSetObjectRecord;
+import org.apache.kafka.common.metadata.UpdateNextNodeIdRecord;
+import org.apache.kafka.metadata.stream.S3ObjectState;
+import org.apache.kafka.metadata.stream.S3StreamSetObject;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+/**
+ * This test is for the extended metadata node manager, only works for AutoMQ.
+ */
+public class ExtMetadataNodeManagerTest {
+    private MetadataNodeManager metadataNodeManager;
+
+    @BeforeEach
+    public void setup() throws Exception {
+        metadataNodeManager = new MetadataNodeManager();
+        metadataNodeManager.setup();
+    }
+
+    @AfterEach
+    public void cleanup() throws Exception {
+        metadataNodeManager.close();
+    }
+
+    @Test
+    public void testS3StreamRecord() {
+        long streamId = 1L;
+        long epoch = 3L;
+        int rangeIndex = 5;
+        long startOffset = 7L;
+        S3StreamRecord record1 = new S3StreamRecord()
+            .setStreamId(streamId)
+            .setEpoch(epoch)
+            .setRangeIndex(rangeIndex)
+            .setStartOffset(startOffset)
+            .setStreamState(StreamState.OPENED.toByte());
+        metadataNodeManager.handleMessage(record1);
+        assertEquals(S3StreamRecordJsonConverter.write(record1, S3StreamRecord.HIGHEST_SUPPORTED_VERSION).toPrettyString(), metadataNodeManager.getData().root().file("s3Streams", Long.toString(streamId), "data").contents());
+
+        RemoveS3StreamRecord record2 = new RemoveS3StreamRecord().setStreamId(streamId);
+        metadataNodeManager.handleMessage(record2);
+        assertFalse(metadataNodeManager.getData().root().directory("s3Streams").children().containsKey(Long.toString(streamId)));
+    }
+
+    @Test
+    public void testRangeRecord() {
+        long streamId = 1L;
+        int rangeIndex = 3;
+        long epoch = 5L;
+        long startOffset = 7L;
+        long endOffset = 9L;
+        int nodeId = 11;
+        RangeRecord record1 = new RangeRecord()
+            .setStreamId(streamId)
+            .setRangeIndex(rangeIndex)
+            .setEpoch(epoch)
+            .setStartOffset(startOffset)
+            .setEndOffset(endOffset)
+            .setNodeId(nodeId);
+        metadataNodeManager.handleMessage(record1);
+        assertEquals(RangeRecordJsonConverter.write(record1, RangeRecord.HIGHEST_SUPPORTED_VERSION).toPrettyString(), metadataNodeManager.getData().root().file("s3Streams", Long.toString(streamId), "ranges", Integer.toString(rangeIndex)).contents());
+
+        RemoveRangeRecord record2 = new RemoveRangeRecord().setStreamId(streamId).setRangeIndex(rangeIndex);
+        metadataNodeManager.handleMessage(record2);
+        assertFalse(metadataNodeManager.getData().root().directory("s3Streams", Long.toString(streamId), "ranges").children().containsKey(Integer.toString(rangeIndex)));
+    }
+
+    @Test
+    public void testS3StreamObjectRecord() {
+        long streamId = 1L;
+        long startOffset = 3L;
+        long endOffset = 5L;
+        long objectId = 7L;
+        long dataTimeInMs = 9L;
+        S3StreamObjectRecord record1 = new S3StreamObjectRecord()
+            .setStreamId(streamId)
+            .setStartOffset(startOffset)
+            .setEndOffset(endOffset)
+            .setObjectId(objectId)
+            .setDataTimeInMs(dataTimeInMs);
+        metadataNodeManager.handleMessage(record1);
+        assertEquals(S3StreamObjectRecordJsonConverter.write(record1, S3StreamObjectRecord.HIGHEST_SUPPORTED_VERSION).toPrettyString(), metadataNodeManager.getData().root().file("s3Streams", Long.toString(streamId), "s3StreamObjects", Long.toString(objectId)).contents());
+
+        RemoveS3StreamObjectRecord record2 = new RemoveS3StreamObjectRecord().setStreamId(streamId).setObjectId(objectId);
+        metadataNodeManager.handleMessage(record2);
+        assertFalse(metadataNodeManager.getData().root().directory("s3Streams", Long.toString(streamId), "s3StreamObjects").children().containsKey(Long.toString(objectId)));
+    }
+
+    @Test
+    public void testS3StreamSetObjectRecord() {
+        int nodeId = 1;
+        long objectId = 3L;
+        long orderId = 5;
+        long dataTimeInMs = 7L;
+
+        List<StreamOffsetRange> rangeList = List.of(
+            new StreamOffsetRange(0, 0L, 100L),
+            new StreamOffsetRange(1, 0L, 200L)
+        );
+        byte[] ranges = S3StreamSetObject.encode(rangeList);
+        S3StreamSetObjectRecord record1 = new S3StreamSetObjectRecord()
+            .setNodeId(nodeId)
+            .setObjectId(objectId)
+            .setOrderId(orderId)
+            .setDataTimeInMs(dataTimeInMs)
+            .setRanges(ranges);
+        metadataNodeManager.handleMessage(record1);
+        assertEquals(Long.toString(orderId), metadataNodeManager.getData().root().file("nodes", Integer.toString(nodeId), "s3StreamSetObjects", Long.toString(objectId), "orderId").contents());
+        assertEquals(Long.toString(dataTimeInMs), metadataNodeManager.getData().root().file("nodes", Integer.toString(nodeId), "s3StreamSetObjects", Long.toString(objectId), "dataTimeInMs").contents());
+
+        RemoveStreamSetObjectRecord record2 = new RemoveStreamSetObjectRecord().setNodeId(nodeId).setObjectId(objectId);
+        metadataNodeManager.handleMessage(record2);
+        assertFalse(metadataNodeManager.getData().root().directory("nodes", Integer.toString(nodeId), "s3StreamSetObjects").children().containsKey(Long.toString(objectId)));
+    }
+
+    @Test
+    public void testS3ObjectRecord() {
+        long objectId = 1L;
+        long objectSize = 3L;
+        long preparedTimeInMs = 5L;
+        long expiredTimeInMs = 7L;
+        long committedTimeInMs = 9L;
+        long markDestroyedTimeInMs = 11L;
+        byte objectState = S3ObjectState.COMMITTED.toByte();
+
+        S3ObjectRecord record1 = new S3ObjectRecord()
+            .setObjectId(objectId)
+            .setObjectSize(objectSize)
+            .setPreparedTimeInMs(preparedTimeInMs)
+            .setExpiredTimeInMs(expiredTimeInMs)
+            .setCommittedTimeInMs(committedTimeInMs)
+            .setMarkDestroyedTimeInMs(markDestroyedTimeInMs)
+            .setObjectState(objectState);
+        metadataNodeManager.handleMessage(record1);
+        assertEquals(S3ObjectRecordJsonConverter.write(record1, S3ObjectRecord.HIGHEST_SUPPORTED_VERSION).toPrettyString(), metadataNodeManager.getData().root().file("s3Objects", Long.toString(objectId)).contents());
+
+        RemoveS3ObjectRecord record2 = new RemoveS3ObjectRecord().setObjectId(objectId);
+        metadataNodeManager.handleMessage(record2);
+        assertFalse(metadataNodeManager.getData().root().directory("s3Objects").children().containsKey(Long.toString(objectId)));
+    }
+
+    @Test
+    public void testAssignedStreamIdRecord() {
+        long streamId = 1L;
+        AssignedStreamIdRecord record1 = new AssignedStreamIdRecord().setAssignedStreamId(streamId);
+        metadataNodeManager.handleMessage(record1);
+        assertEquals(Long.toString(streamId), metadataNodeManager.getData().root().file("s3Streams", "nextStreamId").contents());
+    }
+
+    @Test
+    public void testAssignedS3ObjectIdRecord() {
+        long objectId1 = 1L;
+        AssignedS3ObjectIdRecord record1 = new AssignedS3ObjectIdRecord().setAssignedS3ObjectId(objectId1);
+        metadataNodeManager.handleMessage(record1);
+        assertEquals(Long.toString(objectId1), metadataNodeManager.getData().root().file("s3Objects", "nextObjectId").contents());
+    }
+
+    @Test
+    public void testNodeWalMetadataRecord() {
+        int nodeId = 1;
+        long nodeEpoch = 3L;
+        boolean failoverMode = true;
+        NodeWALMetadataRecord record1 = new NodeWALMetadataRecord()
+            .setNodeId(nodeId)
+            .setNodeEpoch(nodeEpoch)
+            .setFailoverMode(failoverMode);
+        metadataNodeManager.handleMessage(record1);
+        assertEquals(NodeWALMetadataRecordJsonConverter.write(record1, RangeRecord.HIGHEST_SUPPORTED_VERSION).toPrettyString(), metadataNodeManager.getData().root().file("nodes", Integer.toString(nodeId), "walMetadata").contents());
+
+        RemoveNodeWALMetadataRecord record2 = new RemoveNodeWALMetadataRecord().setNodeId(nodeId);
+        metadataNodeManager.handleMessage(record2);
+        assertFalse(metadataNodeManager.getData().root().directory("nodes", Integer.toString(nodeId)).children().containsKey("walMetadata"));
+    }
+
+    @Test
+    public void testKvRecord() {
+        String namespace = "_kafka_FVWxwMClSYObxlUORZANzA";
+        String topicId = "4IHvQLhZSJ6szHPcyady3Q";
+        String partition = "0";
+        String key = namespace + "/" + topicId + "/" + partition;
+        ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+        buffer.putLong(1);
+        byte[] value = buffer.array();
+        KVRecord.KeyValue keyValue = new KVRecord.KeyValue().setKey(key).setValue(value);
+        List<KVRecord.KeyValue> keyValues = List.of(keyValue);
+        KVRecord kvRecord = new KVRecord().setKeyValues(keyValues);
+        metadataNodeManager.handleMessage(kvRecord);
+        assertEquals(Arrays.toString(value), metadataNodeManager.getData().root().file("kvRecords", namespace, topicId, partition).contents());
+
+        RemoveKVRecord removeKVRecord = new RemoveKVRecord().setKeys(List.of(key));
+        metadataNodeManager.handleMessage(removeKVRecord);
+        assertFalse(metadataNodeManager.getData().root().directory("kvRecords", namespace, topicId).children().containsKey(partition));
+    }
+
+    @Test
+    public void testUpdateNextNodeId() {
+        int nodeId = 1;
+        UpdateNextNodeIdRecord record = new UpdateNextNodeIdRecord().setNodeId(1);
+        metadataNodeManager.handleMessage(record);
+        assertEquals(Integer.toString(nodeId), metadataNodeManager.getData().root().file("nodes", "nextNodeId").contents());
+    }
+}