...
Code Block | ||||
---|---|---|---|---|
| ||||
builder.table( "my_topic", Consumed.with(Serdes.Integer(), Serdes.Integer()), Materialized.as(Stores.persistentVersionedKeyValueStore( "my_store", Duration.ofMillis(HISTORY_RETENTION) )) ); final VersionedKeyQuery<Integer, VersionedRecord<Integer>>Integer> query = VersionedKeyQuery.withKey(1).asOf(Instant.parse("2023-08-03T10:37:30.00Z"); final StateQueryRequest<VersionedRecord<Integer>> request = inStore("my_store").withQuery(query); final StateQueryResult<VersionedRecord<Integer>> result = kafkaStreams.query(request); |
...