public byte[] serialize() throws Throwable {
        try (ByteArrayOutputStream outputStream = new ByteArrayOutputStream()) {
            putInt(outputStream, this.replicaInfoTable.size());
            for (Map.Entry<String, BrokerReplicaInfo> entry : replicaInfoTable.entrySet()) {
                final byte[] brokerName = entry.getKey().getBytes(StandardCharsets.UTF_8);
                byte[] brokerReplicaInfo = hessianSerialize(entry.getValue());
                putInt(outputStream, brokerName.length);
                outputStream.write(brokerName);
                putInt(outputStream, brokerReplicaInfo.length);
                outputStream.write(brokerReplicaInfo);
            }
            putInt(outputStream, this.syncStateSetInfoTable.size());
            for (Map.Entry<String, SyncStateInfo> entry : syncStateSetInfoTable.entrySet()) {
                final byte[] brokerName = entry.getKey().getBytes(StandardCharsets.UTF_8);
                byte[] syncStateInfo = hessianSerialize(entry.getValue());
                putInt(outputStream, brokerName.length);
                outputStream.write(brokerName);
                putInt(outputStream, syncStateInfo.length);
                outputStream.write(syncStateInfo);
            }
            return outputStream.toByteArray();
        } catch (Throwable e) {
            LOGGER.error("serialize replicaInfoTable or syncStateSetInfoTable error", e);
            throw e;
        }
    }
@Test
    public void testSerialize() {
        mockMetaData();
        byte[] data;
        try {
            data = this.replicasInfoManager.serialize();
        } catch (Throwable e) {
            throw new RuntimeException(e);
        }
        final ReplicasInfoManager newReplicasInfoManager = new ReplicasInfoManager(config);
        try {
            newReplicasInfoManager.deserializeFrom(data);
        } catch (Throwable e) {
            throw new RuntimeException(e);
        }
        Map<String, BrokerReplicaInfo> oldReplicaInfoTable = new TreeMap<>();
        Map<String, BrokerReplicaInfo> newReplicaInfoTable = new TreeMap<>();
        Map<String/* brokerName */, SyncStateInfo> oldSyncStateTable = new TreeMap<>();
        Map<String/* brokerName */, SyncStateInfo> newSyncStateTable = new TreeMap<>();
        try {
            Field field = ReplicasInfoManager.class.getDeclaredField("replicaInfoTable");
            field.setAccessible(true);
            oldReplicaInfoTable.putAll((Map<String, BrokerReplicaInfo>) field.get(this.replicasInfoManager));
            newReplicaInfoTable.putAll((Map<String, BrokerReplicaInfo>) field.get(newReplicasInfoManager));
            field = ReplicasInfoManager.class.getDeclaredField("syncStateSetInfoTable");
            field.setAccessible(true);
            oldSyncStateTable.putAll((Map<String, SyncStateInfo>) field.get(this.replicasInfoManager));
            newSyncStateTable.putAll((Map<String, SyncStateInfo>) field.get(newReplicasInfoManager));
        } catch (NoSuchFieldException | IllegalAccessException e) {
            throw new RuntimeException(e);
        }
        assertArrayEquals(oldReplicaInfoTable.keySet().toArray(), newReplicaInfoTable.keySet().toArray());
        assertArrayEquals(oldSyncStateTable.keySet().toArray(), newSyncStateTable.keySet().toArray());
        for (String brokerName : oldReplicaInfoTable.keySet()) {
            BrokerReplicaInfo oldReplicaInfo = oldReplicaInfoTable.get(brokerName);
            BrokerReplicaInfo newReplicaInfo = newReplicaInfoTable.get(brokerName);
            Field[] fields = oldReplicaInfo.getClass().getFields();
        }
    }