Kafka Streams：processValues()的NPE及Suppress会话窗口关闭异常

问题1：processValues()触发空指针异常（NPE） #

场景代码 #

@Bean
public Function<KStream<String, String>, KStream<String, String>> process() {
    return inputStream -> inputStream
        .processValues(() -> new HeartbeatProcessor())  // 每秒发送心跳事件
        .groupByKey()
        .windowedBy(SessionWindows.ofInactivityGapAndGrace(
            Duration.ofSeconds(5),  // 非活动间隔
            Duration.ofSeconds(1)   // 宽限期
        ))
        .aggregate(...)
        .suppress(Suppressed.untilWindowCloses(
            Suppressed.BufferConfig.unbounded()
        ));
}

报错信息 #

Caused by: java.lang.NullPointerException: Cannot invoke "String.getBytes(java.nio.charset.Charset)" because "this.topic" is null
   at org.apache.kafka.streams.processor.internals.ProcessorRecordContext.serialize(ProcessorRecordContext.java:97)

注意：用process()替代processValues()时会话窗口可正常关闭，但希望避免重分区。

原因分析 #

processValues()使用的FixedKeyProcessorContext初始化时，默认的ProcessorRecordContext未设置主题名。后续Suppress操作需要序列化记录上下文时，会因topic为null抛出NPE。而process()使用的普通ProcessorContext会继承上游记录的上下文信息，因此不会出现该问题。

解决办法 #

不要强转InternalProcessorContext手动设置主题，正确做法是复用上游记录的上下文。修改HeartbeatProcessor如下：

public class HeartbeatProcessor implements FixedKeyProcessor<String, String, String> {
    private FixedKeyProcessorContext<String, String> context;
    private ProcessorRecordContext recordContext;

    @Override
    public void init(FixedKeyProcessorContext<String, String> context) {
        this.context = context;
        context.schedule(Duration.ofSeconds(1), PunctuationType.WALL_CLOCK_TIME, this::generateHeartbeat);
    }

    @Override
    public void process(FixedKeyRecord<String, String> record) {
        // 保存第一条上游记录的上下文，后续心跳复用
        if (recordContext == null) {
            recordContext = ((InternalFixedKeyRecord) record).context();
        }
        context.forward(record);
    }

    private void generateHeartbeat(long timestamp) {
        if (recordContext == null) {
            return; // 未收到上游记录时不发送心跳
        }
        
        // 创建带有效上下文的心跳记录
        InternalFixedKeyRecord<String, String> heartbeatRecord = InternalFixedKeyRecordFactory.create(
            recordContext.withTimestamp(timestamp),
            "heartbeat-" + timestamp,
            "heartbeat"
        );
        context.forward(heartbeatRecord);
    }
}

问题2：修复NPE后Suppress的会话窗口无法正常关闭 #

场景代码 #

用户手动设置主题名修复NPE，但窗口关闭逻辑异常，代码如下：

public class HeartbeatProcessor implements FixedKeyProcessor<String, String, String> {
    private FixedKeyProcessorContext<String, String> context;

    @Override
    public void init(FixedKeyProcessorContext<String, String> context) {
        this.context = context;
        context.schedule(Duration.ofSeconds(1), PunctuationType.WALL_CLOCK_TIME, this::generateHeartbeat);
    }

    @Override
    public void process(FixedKeyRecord<String, String> record) {
        context.forward(record);
    }

    private void generateHeartbeat(long timestamp) {
        if (context instanceof InternalProcessorContext internalContext) {
            internalContext.setRecordContext(
                new ProcessorRecordContext(
                    timestamp,
                    0L,
                    context.taskId().partition(),
                    "dummy-topic",
                    new RecordHeaders()
                )
            );
        }
        
        Record<String, String> record = new Record<>(
            "heartbeat-" + timestamp,  // 每秒生成不同的Key
            "heartbeat",
            timestamp
        );
        context.forward(InternalFixedKeyRecordFactory.create(record));
    }
}

原因分析 #

1. 心跳Key错误：生成的心跳Key为"heartbeat-" + timestamp，每个心跳都属于独立会话窗口，无法触发原有业务Key的窗口关闭（会话窗口按Key分组，不同Key的心跳不影响目标窗口生命周期）。
2. 上下文不匹配：手动创建的ProcessorRecordContext未关联业务Key的分区和窗口元数据，Suppress操作无法正确判断窗口是否该关闭。

解决办法 #

方案1：针对业务Key发送心跳（推荐）

FixedKeyProcessor绑定单个Key，心跳需使用相同业务Key才能触发对应窗口的非活动检测：

private void generateHeartbeat(long timestamp) {
    if (recordContext == null) {
        return;
    }
    // 复用当前处理器绑定的业务Key
    InternalFixedKeyRecord<String, String> heartbeatRecord = InternalFixedKeyRecordFactory.create(
        recordContext.withTimestamp(timestamp),
        recordContext.key(),
        "heartbeat"
    );
    context.forward(heartbeatRecord);
}

方案2：调整Suppress与窗口参数

若需全局心跳，可修改Suppress配置明确窗口关闭时间：

.suppress(Suppressed.untilTimeLimit(
    Duration.ofSeconds(6), // 非活动间隔+宽限期
    Suppressed.BufferConfig.unbounded()
))

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