@EmbeddedKafka not working properly with GlobalKTable and transformer

@ecristobal commented on Mon Feb 07 2022

Hi guys!

In this repo I've attached a project that is not currently working. To sum up what it is intended to do, we have 2 input topics: one with incoming purchases (KStream) and one with an item catalog (GlobalKTable, materialized as a state store). The idea is, each time we receive a purchase, we enrich its content with a property stored on the KTable, and with some other operations an enriched purchase object will be returned. There are also a couple of caches implemented with Spring and Redis to cache calls to a couple of external methods (I've added the code to give you full context, but I don't think they affect tp the result of the sample).

However, I didn't find a way to make this example work. I've tried to switch from GlobalKTable to KTable, follow your documentation about state stores, transformations and so on, but it wasn't possible to me to make this work. The project also contains a sample test to reproduce the problem here.

Please, can you help us in figuring out what is happening?

Thanks and regards!

Start with the linked StreamFunctionIT.java reproduction test and the example repository to understand the @EmbeddedKafka setup, GlobalKTable state store, and transformer flow. Determine why the sample does not work and verify the fix by getting the integration test to pass with the purchase enrichment behavior described.