使用MockProcessorContext做Kafka Streams单元测试遇sensor空指针异常

你遇到的NullPointerException是因为手动构建的MeteredKeyValueStore没有关联有效的StreamsMetrics实例，导致内部的sensor字段为null，执行put操作时触发空指针。

修改添加状态存储的代码，直接将StoreBuilder传入MockProcessorContext.addStateStore()，而非先调用build()方法。MockProcessorContext会自动为Store绑定正确的Metrics实例，避免空指针问题。

修正后的代码：

val props = Properties()
props[StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG] = Serdes.String().javaClass
props[StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG] = Serdes.String().javaClass

val processorUnderTest = MyProcessor("test-store-name", keySerde, valueSerde)
val context = MockProcessorContext(props)     

// 直接传入StoreBuilder，不调用build()
context.addStateStore(
    Stores.keyValueStoreBuilder(
        Stores.inMemoryKeyValueStore("test-store-name"),
        Serdes.String(),
        Serdes.String()
    )
)

// 初始化处理器
processorUnderTest.init(context)

如果MyProcessor在初始化阶段会直接访问状态存储，必须确保先通过context.addStateStore()完成Store添加，再调用processorUnderTest.init(context)，保证处理器能获取到正确初始化的状态存储实例。

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