Can I Emit to a KTable?

[pfortin-urbn]

@frairon - I am in a processor and it has received a message to update the KTable but it could be for another key that the context key. Can I use ctx.Emit to update the KTable and if I do that will local storage also be updated? What is the preferred way to update the state of a key that is determined in the processor?

[jomaresch]

Hey @pfortin-urbn.

I think the ctx.Loopback() function can solve this.

goka/context.go

Line 117 in 15746b5

Loopback(key string, value interface{}, options ...ContextOption)

So if you receive a message with the key key_a and you want to update the stored value of key_b you can loopback the message to key_b using ctx.Loopback("key_b", updateMessage).
In the Loop-Edge you define how to update the value.

goka/graph.go

Line 324 in 0085fa7

func Loop(c Codec, cb ProcessCallback) Edge {

Here is also a good example.
Based on one input message, we update the values of 2 different keys (the receiver_id and sender_id).

I hope this helps 🙂

[db7]

By KTable do you mean a KTable of Kafka Streams? If yes, you'll need to configure the hash function since Sarama uses a different hash function than the Java clients. You'll need to create an emitter, configure the hash function (murmur2 AFAIR) and pass the emitter object to the process callback.

[jomaresch]

Oh sorry for the confusion. Somehow i totally missed the KTable part.

[pfortin-urbn]

@db7 Thanks I kinda knew that but wanted to know if there was easier way thanks!