spring-cloud / spring-cloud/spring-cloud-stream
I want to consume KTable and materialize with optimization
Nobody has claimed this yet.
- Dominant language
- Java
- Stars
- 1.1k
- Forks
- 646
- Avg merge
- 2d 3h
- Merged PRs (30d)
- 8
Description
When I consume KTable, it is materialized as KeyValueStore automatically.
fun process() = BiConsumer<KStream<String, String>, KTable<String, String>> { input, table ->
input.join(table) { value1, value2 ->
Pair(value1, value2)
}.peek { key, value -> println("$key: $value") }
}
If I set topology.optimization to all then, changelog topic is not created but reuse consuming topic as changelog topic.
Because Kafka Streams join semantics, I want to change the type of state store to VersionedKeyValueStore.
If I materialize state store manually like this, it creates changelog topic although I set topology.optimization to all.
fun process() = BiConsumer<KStream<String, String>, KTable<String, String>> { input, table ->
val storedTable = table
.toStream()
.groupByKey()
.aggregate(
{ byteArrayOf() },
{ _, value, _ -> value.toByteArray() },
Materialized.`as`(store)
)
val store = Stores.persistentVersionedKeyValueStore("kafka-streams-test-store", Duration.ofDays(1))
val stream = input.join(storedTable) { value1, value2 ->
Pair(value1, value2)
}.peek { key, value -> println("$key, ${value?.first}, ${value?.second?.decodeToString()") }
}
Is there any way to change the type of state store with not creating changelog topic via optimization?
Contributor guide
No contributing guide indexed for this repository
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start with the KTable join and materialization paths shown in the issue, then trace how topology.optimization=all handles the automatically materialized store versus the explicit VersionedKeyValueStore. The issue does not name source files or tests; done would require a documented or tested way to select the store type without creating an unnecessary changelog topic.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- kafka, kotlin
- Domain
- stream-processing
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100