Облачная функция Spring для создания GlobalKTable из strem

Question

Есть ли пример того, как создать GlobalKTable, чтобы вести счет от KStream, используя поток Spring Cloud и используя функциональный подход?

Answer 1

Является ли реализация интерфейса процессора правильным подходом?

    @Bean
    public Consumer<KStream<String, Long>> processorsample() {
        return input -> input.process(() -> new Processor<String, Long>() {

            @Override
            public void init(ProcessorContext context) {
                if (state == null) {
                    state = (KeyValueStore<String, Long>) context.getStateStore("mystate");
                } 
            }

            @Override
            public void process(String key, Long value) {
                if (state != null) {
                    if (key != null) {
                        Long currentCount = state.get(key);
                        if (currentCount == null) {
                            state.put(key, value);
                        } else {
                            state.put(key, currentCount + value);
                        }

                    }

                }
            }

            @Override
            public void close() {
                if (state != null) {
                    state.close();
                }
            }
        }, "mystate");

    }