Hi folks I’m using

.connect()

for an enrichment use case much like what is described here. I’m doing a few subsequent .connects() to enrich with a few debezium tables. What is surprising is I seem to get a single record output to Kafka for for the first join match but nothing else ever gets written. Basically I just want to enrich a steady stream of data with some meta data that sometimes changes but not that often. I would have hoped in my diagram I would see 236 records output instead of just 1. Is this not a good example of how to use .connect() or is the pattern like what is shown in the

RichCoFlatMapFunction

not the one to follow if I want subsequent joins to produce a result? I feel like I should be describing my input stream “encodedproto” in some different way that tells Flink to keep enriching every record. It sort of seems like the join matches on the first record and then nothing gets updated after that. Any help / pointers to docs or examples much appreciated!

The join implemented in that Rides and Fares example is designed specifically for the case where the two streams are perfectly paired — there’s exactly one fare for each ride, and these pairs are joined together. It sounds like you have different join semantics, so you’ll need to adjust the implementation accordingly. In particular, I believe you don’t want to immediately clear the ride and fare after the first join result is found. FWIW, for most use cases, the Table and SQL APIs make it much easier to implement enrichment joins.

Thanks David. I had actually started with a 5-line SQL query that seemed to work great. However when I tried to write the output to a Kafka topic I got this error, and the person who helped me said I should use the data stream API (didn’t get too many details and the example I found was from Googling). I’m relatively new to Flink and flexible to use either. I just want to write an enrichment join to a different Kafka topic. Could you point me to an example of either table or datastream API that does what I want? I’ll experiment with removing the clear. I thought since the out.collect(..) was getting called (verified with some debug logs) that is what sent the message to Kafka / the output https://apache-flink.slack.com/archives/C03G7LJTS2G/p1691787879203899?thread_ts=1691787063.492379&cid=C03G7LJTS2G