我可以通过创建一个StoreBuilder bean来实现：

@Bean
public StoreBuilder<KeyValueStore<String, String>> storeBuilder() {
    return Stores.keyValueStoreBuilder(
            Stores.persistentKeyValueStore("value-store"),
            Serdes.String(),
            Serdes.String()
    );
}

然后使用KStream的这个store in transform方法如下：

@Bean
public Function<KStream<String, String>, KStream<String, Long>> sampleProcessor() {
    return source -> source
            .transform(() -> new Transformer<String, String, KeyValue<String, String>>() {
                private KeyValueStore<String, String> valueStore;

                @Override
                public void init(ProcessorContext context) {
                    valueStore = context.getStateStore("value-store");
                }

                @Override
                public KeyValue<String, String> transform(String key, String value) {
                    if (isNull(valueStore.get(value))) {
                        valueStore.put(value, "CONST");
                        return KeyValue.pair(key, value);
                    }
                    return null;
                }

                @Override
                public void close() {
                }
            }, "value-store")
            .mapValues(value -> value.toUpperCase())
            .groupBy((key, value) -> value)
            .count(Materialized.as("count-store"))
            .toStream();
}