Resolving Hazelcast ReplicatedMap ClassCastException with InMemoryFormat.BINARY

java.lang.ClassCastException: class java.lang.String cannot be cast to class com.hazelcast.internal.serialization.impl.HeapData

if (isBinary) {
  memoryUsage += ((HeapData) record.getValueInternal()).getHeapCost(); // <-- 异常发生在此处
}

ReplicatedMap<String, String> map = hz.getReplicatedMap("rogueUsers");

ReplicatedMap<Data, Data> map = hz.getReplicatedMap("rogueUsers");

import com.hazelcast.config.Config;
import com.hazelcast.config.InMemoryFormat;
import com.hazelcast.config.JoinConfig;
import com.hazelcast.config.NetworkConfig;
import com.hazelcast.config.ReplicatedMapConfig;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.internal.serialization.Data; // 引入Data类
import com.hazelcast.map.listener.EntryAddedListener;
import com.hazelcast.map.listener.EntryRemovedListener;
import com.hazelcast.map.listener.EntryUpdatedListener;
import com.hazelcast.replicatedmap.ReplicatedMap;
import com.hazelcast.core.EntryEvent;

import java.io.Serializable;

public class HazelcastConfigSetup {

    private static HazelcastInstance setupHazelcastConfig() {
        Config config = new Config();
        config.setInstanceName("rogueUsers");
        NetworkConfig network = config.getNetworkConfig();
        network.setPort(5701).setPortCount(20);
        network.setPortAutoIncrement(true);
        JoinConfig join = network.getJoin();
        join.getMulticastConfig().setEnabled(true);
        // join.getTcpIpConfig().setEnabled(true); // 如果使用TCP/IP，取消注释

        HazelcastInstance hz = Hazelcast.getOrCreateHazelcastInstance(config);

        ReplicatedMapConfig replicatedMapConfig =
                config.getReplicatedMapConfig("rogueUsers");

        replicatedMapConfig.setInMemoryFormat(InMemoryFormat.BINARY);
        replicatedMapConfig.setAsyncFillup(true);
        replicatedMapConfig.setStatisticsEnabled(true);
        replicatedMapConfig.setSplitBrainProtectionName("splitbrainprotection-name");

        // 核心修改：将ReplicatedMap的泛型类型改为Data, Data
        ReplicatedMap<Data, Data> map = hz.getReplicatedMap("rogueUsers");
        map.addEntryListener(new RogueEntryListener());

        return hz;
    }

    // 示例EntryListener，需要根据实际业务逻辑调整
    // 注意：如果ReplicatedMap的泛型改为Data，那么EntryListener中的EntryEvent也会是Data类型
    // 在实际使用中，你需要手动对Data进行反序列化
    static class RogueEntryListener implements EntryAddedListener<Data, Data>,
            EntryUpdatedListener<Data, Data>, EntryRemovedListener<Data, Data>, Serializable {

        @Override
        public void entryAdded(EntryEvent<Data, Data> event) {
            System.out.println("Entry Added: " + event.getKey() + " -> " + event.getValue());
            // 示例：如何从Data对象获取原始值
            // String keyStr = (String) event.getOldValue().getSerializationService().toObject(event.getKey());
            // String valueStr = (String) event.getOldValue().getSerializationService().toObject(event.getValue());
        }

        @Override
        public void entryRemoved(EntryEvent<Data, Data> event) {
            System.out.println("Entry Removed: " + event.getKey() + " -> " + event.getOldValue());
        }

        @Override
        public void entryUpdated(EntryEvent<Data, Data> event) {
            System.out.println("Entry Updated: " + event.getKey() + " -> " + event.getOldValue() + " to " + event.getValue());
        }
    }

    public static void main(String[] args) {
        HazelcastInstance hz = setupHazelcastConfig();
        ReplicatedMap<Data, Data> map = hz.getReplicatedMap("rogueUsers");

        // 示例：如何存入和取出数据
        String originalKey = "user:123";
        String originalValue = "status:active";

        // 将String序列化为Data对象存入Map
        Data keyData = hz.getSerializationService().toData(originalKey);
        Data valueData = hz.getSerializationService().toData(originalValue);
        map.put(keyData, valueData);
        System.out.println("Put: " + originalKey + " -> " + originalValue);

        // 从Map中取出Data对象并反序列化为String
        Data retrievedValueData = map.get(keyData);
        if (retrievedValueData != null) {
            String retrievedValue = hz.getSerializationService().toObject(retrievedValueData);
            System.out.println("Get: " + originalKey + " -> " + retrievedValue);
        }

        hz.shutdown();
    }
}