Hi,
I'm definitly no expert, but to me it sounds as not all your messages are
getting processed by the same consumer. Are you using the key `foo` for
partitioning? Is `baz` actually another key or is this mixup in your example
and `baz` is another value with key `foo`?
Hope you find a solution to your problem.
Best,
Claudia
Von: Mauricio Lopez
Gesendet: Donnerstag, 17. August 2023 22:57
An: users@kafka.apache.org
Betreff: Table updates are not consistent when doing a join with a Stream
Hello Folks,
We are having an issue with a Kafka Streams Java application.
We have a KStream and a KTable which are joined using a Left Join. The entries
in the KTable are constantly updated by the new information that comes from the
KStream. Each KStream message is adding entries to an array that the KTable has
for each key. This update gets sent back to the KTable topic, expanding this
array every time a new message comes from the KStream.
As an example, what should be happening (and what happens in our unit tests) is:
* KTable has an empty array for key “foo”: []
* Event 1 comes with key “foo” and value “bar”
* Ktable gets updated to “foo”: [“bar”] , sending this update´to´ the same
topic that the KTable is plugged into.
* Event 2 comes with key “baz”
* Update is pulled to mem by Ktable, and the Ktable gets updated to “foo”:
[“bar, “baz”], sending this change ´to´ the same topic that the KTable is
plugged into. Baz was appended to the array for key “foo”.
But what is happening is the following:
* KTable has an empty array for key “foo”: []
* Event 1 comes with key “foo” and value “bar”
* Ktable gets updated to “foo”: [“bar”] in the joiner, sending an event
´to´ the same topic that the KTable is plugged to.
* Event 2 comes with key “baz”
* Ktable gets updated to “foo”: [“baz”] in the joiner, sending an event
´to´ the same topic that the KTable is plugged to afterwards.
This happens multiple times, and after a couple of seconds, one of the incoming
messages is finally appended, but many of them are lost. As you can see, we
suspect that when the Event 2 is received, the KTable has somehow not received
the first update for adding “baz” to the array.
This means that many events are missed, and we cannot successfully get the
KTable to save all the data for all the events. In turn, it sometimes
overwrites the updates from some events.
So far, we have tried:
* Setting STATESTORE_CACHE_MAX_BYTES_CONFIG to 0, to attempt to force the
app not to cache any changes and send to the output topic instantly.
* Setting COMMIT_INTERVAL_MS_CONFIG to 0, to attempt to force the app to
send all updates instantly
* Setting TOPOLOGY_OPTIMIZATION_CONFIG to “reuse.ktable.source.topics” and
“all” in case there is some optimization pattern that could help us.
None of these have allowed us to have a fully consistent update of the KTable
each time a new event comes. It always gets overwritten or misses incoming
updates made by events. Can someone advice if there’s a way to make the KTable
get successfully updated by each one of the events, as the first example shows?
Thanks,
Mauricio L
This message is for the sole use of the intended recipient(s) and may contain
confidential and/or privileged information of SilverSky. Any unauthorized
review, use, copying, disclosure, or distribution is prohibited. If you are not
the intended recipient, please immediately contact the sender by reply email
and delete all copies of the original message.
