Kafka自定义状态存储未更新排查：映射始终为空

你遇到的核心问题是修改状态存储中的Map对象后，没有将更新后的Map重新写回存储——Kafka Streams的状态存储不会自动追踪内存中对象的修改，你必须显式调用put方法把更新后的对象持久化回去。另外还有几处潜在的空指针风险，我一步步给你拆解：

1. 连接事件（CONNECTED）分支的核心错误 #

看你这段代码：

if(stateStore.get(key) == null) {
    stateStore.put(key, new HashMap<>());
    sendUpdate = true;
}
stateStore.get(key).put(value.getSessionId(), value);

你创建新Map并写入存储后，又读取出来添加元素，但这个添加操作只在内存中的临时Map实例里生效，没有把更新后的Map重新存回状态存储。当下次读取这个key时，拿到的还是最初空的Map（或是序列化/反序列化后生成的新空实例），所以看起来大小始终为0。

修复后的代码：

Map<String, ConnectionEvent> userSession = stateStore.get(key);
boolean sendUpdate = false;
if (userSession == null) {
    userSession = new HashMap<>();
    sendUpdate = true;
}
userSession.put(value.getSessionId(), value);
// 关键：把更新后的Map写回状态存储
stateStore.put(key, userSession);

2. 断开事件（DISCONNECTED）分支的双重问题 #

这里有两个明显的问题：

• 没有检查stateStore.get(key)是否为null，直接调用remove会触发空指针异常
• 修改Map后同样没有写回存储，导致存储里的内容始终不更新

修复后的代码：

Map<String, ConnectionEvent> userSession = stateStore.get(key);
if (userSession != null) {
    userSession.remove(value.getSessionId());
    // 关键：写回更新后的Map
    stateStore.put(key, userSession);
    if (userSession.isEmpty()) {
        // 会话为空时可以删除key节省存储
        stateStore.delete(key);
        return KeyValue.pair(key, "Offline");
    }
}
// 会话不存在或还有其他连接时，不发送下游更新
return null;

3. 额外的优化建议 #

• 移除Transformer中冗余的Serde定义：你在Transformer里重复创建了userSessionsSerde，但状态存储的Serde已经在StoreBuilder中配置完成，这里完全不需要重复定义。
• 严格判空防护：所有从状态存储获取的对象都要先做null检查，避免运行时异常。

完整修复后的Transformer代码 #

public class WSEventProcessor implements Transformer<String, ConnectionEvent, KeyValue<String, String>> {
    private static final String STORE_NAME = "user-sessions";
    private KeyValueStore<String, Map<String, ConnectionEvent>> stateStore;
    private ProcessorContext context;

    @SuppressWarnings("unchecked")
    @Override
    public void init(ProcessorContext context) {
        this.context = context;
        stateStore = (KeyValueStore<String, Map<String, ConnectionEvent>>) context.getStateStore(STORE_NAME);
    }

    @Override
    public void close() {
        // 按需添加资源清理逻辑
    }

    @Override
    public KeyValue<String, String> transform(String key, ConnectionEvent value) {
        if (value.getState() == WebSocketConnection.CONNECTED) {
            boolean sendUpdate = false;
            Map<String, ConnectionEvent> userSession = stateStore.get(key);
            if (userSession == null) {
                userSession = new HashMap<>();
                sendUpdate = true;
            }
            userSession.put(value.getSessionId(), value);
            stateStore.put(key, userSession);
            return sendUpdate ? KeyValue.pair(key, "Online") : null;
        } else {
            Map<String, ConnectionEvent> userSession = stateStore.get(key);
            if (userSession != null) {
                userSession.remove(value.getSessionId());
                stateStore.put(key, userSession);
                if (userSession.isEmpty()) {
                    stateStore.delete(key);
                    return KeyValue.pair(key, "Offline");
                }
            }
            return null;
        }
    }
}

这样修改后，状态存储里的Map就会正确保存更新后的内容，你就不会再看到大小始终为0的问题了。

内容的提问来源于stack exchange，提问作者cppcoder