Thanks Matthias, it makes sense, now I need to find out why the topic is not sorted by timestamp.
The topic I'm using to be loaded as globalKTable is partitioned by key, and the timestamp is being handled by kafka. I have multiple clients, different machines, pushing data to that topic, maybe the problem is a time synchronization issue between machines. Do you know if the timestamp associated with each record on topic is being set up by clients pushing the data, or by the kafka broker leader of that partition? Thanks.- On Thu, Jul 9, 2020 at 4:45 PM Matthias J. Sax <mj...@apache.org> wrote: > If you load data into a KTable or GlobalKTable, it's expected that the > data is partitioned by key, and that records with the same key have > non-descending timestamps. > > If a record with let's say key A and timestamp 5 is put into the table, > and later a record with key A and timestamp 4 is put into the table, the > second record is an out-of-order record and a warning is logged for this > case, as it might indicate that your data is no ordered correctly what > might lead to incorrect/unexpected join results. > > If data is out-of-order your table content semantically goes (partially) > back in time what is usually undesired -- your table content should > usually go forward in time only. > > Does this help? > > > -Matthias > > > > On 7/8/20 10:54 AM, Richard Rossel wrote: > > Hi there, > > > > I'm getting lot of this type of warning: > > > > > > WARN org.apache.kafka.streams.kstream.internals.KTableSource - Detected > > out-of-order KTable update for entity-STATE-STORE-0000000000 at offset > > 65806, partition 5. > > > > It looks like the warning is generated each time a new record goes into > the > > source topic which is being used to load a globalKTable, and only happens > > when the source topic is updated very frequently (each second). > > > > I would like to know what is the meaning of that warning, maybe I'm not > > loading the Ktable correctly. > > > > My process is basically a simple enrichment process (Using Kafka 2.4): > > ... > > > > GlobalKTable<String, String> entity = builder.globalTable(entityTopic); > > > > KStream<String, String> joined = dataStream.join(entity, > > > > (leftKey, leftValue) -> leftKey, > > > > joiner); > > > > joined.to(enrichedTopic); > > ... > > > > The entityTopic and the topic used by dataStream, are sharing the same > key > > type (username), > > and the entityTopic is receiving entries from another process very > > frequently. > > > > Do you think those warning messages are because of the way I'm creating > the > > Ktable? > > > > Thanks.- > > > > -- Richard Rossel Atlanta - GA
