如何从KStream获取KafkaStream的状态

import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Transformer;
import org.apache.kafka.streams.processor.ProcessorContext;

public class MyTransformer implements Transformer<KeyType, ValueType, TransformedValueType> {

    private ProcessorContext context;

    @Override
    public void init(ProcessorContext context) {
        this.context = context;
    }

    @Override
    public TransformedValueType transform(KeyType key, ValueType value) {
        // 获取当前状态
        StateStore stateStore = context.getStateStore("state-store");
        StateValue currentState = stateStore.get(key);

        // 更新状态
        StateValue newState = updateState(currentState, value);
        stateStore.put(key, newState);

        // 返回转换后的记录
        return transformValue(value, newState);
    }

    @Override
    public void close() {
        // 清理资源
    }
}

// 在Kafka Streams应用程序中使用Transformer
KStream<KeyType, ValueType> inputStream = builder.stream("input-topic");
KStream<KeyType, TransformedValueType> outputStream = inputStream.transform(
    () -> new MyTransformer(),
    "state-store"
);