Оконный магазин не заполняется (Spring Cloud Stream Kafka) - PullRequest
Новая
       43

Оконный магазин не заполняется (Spring Cloud Stream Kafka)

1 голос
/ 09 мая 2019

Я пытаюсь создать оконное хранилище из набора точек данных, но похоже, что ветвь топологии потока по какой-то причине не оценивается.

Я использую тот же поток для заполнения KTable материализованного в магазине, и это прекрасно работает.

Я использую Spring Cloud Streams со следующей конфигурацией: 

spring:
  application.name: stream-test

  kafka.bootstrap-servers: localhost:9092

  cloud.stream:
    # assign group and topic name to binding
    bindings:
      windowedStream:
        destination: myTopic
        group: stream-test-window
    kafka:
      # configure kafka binder
      binder:
        brokers: ${spring.kafka.bootstrap-servers}
        configuration.auto.offset.reset: latest
      # kafka-streams specific binding configuration
      streams.bindings.windowedStream.consumer:
        keySerde: org.apache.kafka.common.serialization.Serdes$StringSerde
        valueSerde: kstreamstest.StreamSerdes$DataItemSerde

DataItemSerde - это просто расширенный JSON-Serde (также работает для KTable). 

@Data class DataItem {
    String value;
}

public class StreamSerdes {
    public static final Serde<DataItem> DATA_ITEM_SERDE = new DataItemSerde();
    public static class DataItemSerde extends JsonSerde<DataItem> {}
}

с привязкой 

interface WindowedTableBinding {
    String WINDOW_STREAM = "windowedStream";

    @Input(WINDOW_STREAM)
    KStream<String, DataItem> stream();
}

Я создаю потоковый слушатель вот так 

@Configuration
@EnableBinding(WindowedTableBinding.class)
class StreamToWindowed {
    String storeName = "wvs";

    @Bean
    String windowedStoreName() {
        return storeName;
    }

    @StreamListener(WindowedTableBinding.WINDOW_STREAM)
    public void windowStream(@Input(WindowedTableBinding.WINDOW_STREAM) KStream<String, DataItem> stream) {
        stream.peek((k, v)  -> System.out.printf("%s: %s%n", k, v))
                .groupByKey()
                .windowedBy(TimeWindows.of(5_000))
                .reduce((d1, d2) -> d2,
                    Materialized
                        .<String, DataItem, WindowStore<Bytes, byte[]>>as("wvs")
                        .withKeySerde(Serdes.String())
                        .withValueSerde(StreamSerdes.DATA_ITEM_SERDE));
    }
}

Однако, когда я затем запрашиваю магазин 

Set<String> getWindowedKeys() {
    ReadOnlyWindowStore<String, DataItem> queryableStore = queryService
            .getQueryableStore(windowedStoreName, QueryableStoreTypes.windowStore());
    Set<String> result = new HashSet<>();
    if (queryableStore != null) { // store is not null though
        try (KeyValueIterator<Windowed<String>, DataItem> values = queryableStore.all()) {
            values.forEachRemaining(kvs -> result.add(kvs.key.key()));
        }
    }
    return result;
}

этот набор всегда пуст (после отправки данных, конечно). Операторы System.out.print также не запускаются, поэтому, я думаю, ветвь вообще не оценивается.

Опять же, я устанавливаю KTable для тех же значений параллельно, и это прекрасно заполняется. Я могу удалить его, и оконная версия все еще не работает.

Я видел этот пример , но я вижу только одно отличие - он записывает данные обратно в выходной поток, чего я не хочу делать. Кроме того, это не поможет, если я добавлю это.

я тоже пробовал 

@KafkaStreamsStateStore(name="wvs", type= KafkaStreamsStateStoreProperties.StoreType.WINDOW, lengthMs=5_000)
public void windowStream(@Input(WindowedTableBinding.WINDOW_STREAM) KStream<String, DataItem> stream) {}

но это не имеет значения.

Что мне нужно сделать, чтобы заполнить это оконное хранилище данных?

1 Ответ

1 голос
/ 09 мая 2019

Как обычно для меня с Spring, это оказалось проблемой конфигурации.

Мне нужно было дать отдельные application-id s для двух привязок: 

spring.cloud.stream.kafka.streams.bindings:
    tableStream.consumer:
        application-id: table-generator
    windowedStream.consumer:
        application-id: windows-generator
...