Dfyjkt

We have a kTable where we map the values and materialize to KeyValueStore using the following code:

 @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"));

 

And mapper code:

 @Override
 public CancelEvent apply(CancelEvent value) 

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

 return value;

 

We use Avro for serialization and deserialization. When we try to query the state store with a key, it always returns null.

Code for querying the state store:

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

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

event is always null. While debugging, I iterated through the keyValuestore, and realized something is appened to the key (magic byte may be!). Please refer the image below

So, the question is how to query with key?

Update 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;