Kafka State Store возвращает ноль при использовании Avro - PullRequest
Новая
       23

Kafka State Store возвращает ноль при использовании Avro

0 голосов
/ 08 ноября 2018

У нас есть таблица kTable, в которой мы отображаем значения и материализуем их в KeyValueStore, используя следующий код: 

    @Bean
    public KTable<String, CancelEvent> kTable(StreamsBuilder kStreamBuilder,
                                          ValueMapper<CancelEvent, CancelEvent> mapper,
                                          SpecificAvroSerde<CancelEvent> serde) {

      return kStreamBuilder
            .table(properties.getProperty("topics.cancel"), Consumed.with(Serdes.String(), serde))
            .mapValues(mapper, Materialized.as("cancel-event-store"));

    }

и код картографа: 

    @Override
    public CancelEvent apply(CancelEvent value) {

      if (value.getData().getCancelled()) {
        return null;
      }

      return value;

    }

Мы используем Avro для сериализации и десериализации. Когда мы пытаемся запросить ключ state store, он всегда возвращает null.

Код для запроса государственного хранилища: 

    ReadOnlyKeyValueStore<String, CancelEvent> store = streamsFactory.getKafkaStreams()
            .store("cancel-event-store", QueryableStoreTypes.keyValueStore());

    CancelEvent event = store.get(queryEvent.getMeta().getId().toString());

event всегда null. Во время отладки я перебрал keyValuestore и понял, что к ключу что-то добавлено (может быть магический байт!). Пожалуйста, обратитесь к изображению ниже

enter image description here

Итак, вопрос в том, как запросить ключ?

Обновление 1 

    private Map<String, Object> kStreamsConfigsProperties() {
    final Map<String, Object> config = new HashMap<>();

    config.put(APPLICATION_ID_CONFIG, "app.name" + new Date().getTime());

    config.put(BOOTSTRAP_SERVERS_CONFIG, properties.getProperty("spring.kafka.bootstrap-servers"));
    config.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, properties.getProperty("spring.kafka.schema-registry"));

    config.put(DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    config.put(DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class.getName());
    config.put(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, WallclockTimestampExtractor.class.getName());

    return config;
}
...