I'm wondering if the kafka architects have plans to redesign/enhance this behavior. Having to guess the idle time isn't the most satisfying solution. No matter what time I put in there it seems possible that I will miss a join.
Respectfully, Chad On Fri, Nov 5, 2021 at 3:07 PM Matthias J. Sax <mj...@apache.org> wrote: > The log clearly indicates that you hit enforced processing. We record > the metric and log: > > Cf > > https://github.com/apache/kafka/blob/3.0.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGroup.java#L194-L200 > > Not sure why the metric does not report it... > > Hence, the solution would be to increase `max.task.idle.ms` further to > give Kafka Streams time to fetch the data. > > If might help to use DEBUG log to see for which partitions the consumer > sends fetch requests and which partitions return data, to better > understand the underlying behavior. > > > -Matthias > > On 11/5/21 6:58 AM, Chad Preisler wrote: > > It seems like I have 2 options to work around this issue. > > > > > > - Keep the KTable and have another process running that puts the > missed > > join message back on the event topic. > > - Switch to GlobalKTable. > > > > Any other solutions/workarounds are welcome. > > > > Thanks, > > Chad > > > > On Thu, Nov 4, 2021 at 11:43 AM Chad Preisler <chad.preis...@gmail.com> > > wrote: > > > >> enforced-processing-total is zero for all missed join occurrences. I > >> logged all the metrics out at the time my stream processed the missed > join, > >> so let me know if there are any other metics that would help. > >> > >> On Wed, Nov 3, 2021 at 9:21 PM Chad Preisler <chad.preis...@gmail.com> > >> wrote: > >> > >>> I'm not sure. When I ran with trace logging turned on I saw a bunch of > >>> messages like the ones below. Do those messages indicate > >>> "enforced-processing"? It gets logged right after the call > >>> to enforcedProcessingSensor.record. > >>> > >>> Continuing to process although some partitions are empty on the broker. > >>> There may be out-of-order processing for this task as a result. > Partitions > >>> with local data: [status-5]. Partitions we gave up waiting for, with > their > >>> corresponding deadlines: {event-5=1635881287722}. Configured > >>> max.task.idle.ms: 2000. Current wall-clock time: 1635881287750. > >>> > >>> Continuing to process although some partitions are empty on the broker. > >>> There may be out-of-order processing for this task as a result. > Partitions > >>> with local data: [event-5]. Partitions we gave up waiting for, with > their > >>> corresponding deadlines: {status-5=1635881272754}. Configured > >>> max.task.idle.ms: 2000. Current wall-clock time: 1635881277998. > >>> > >>> On Wed, Nov 3, 2021 at 6:11 PM Matthias J. Sax <mj...@apache.org> > wrote: > >>> > >>>> Can you check if the program ever does "enforced processing", ie, > >>>> `max.task.idle.ms` passed, and we process despite an empty input > buffer. > >>>> > >>>> Cf > https://kafka.apache.org/documentation/#kafka_streams_task_monitoring > >>>> > >>>> As long as there is input data, we should never do "enforced > processing" > >>>> and the metric should stay at zero. > >>>> > >>>> > >>>> -Matthias > >>>> > >>>> On 11/3/21 2:41 PM, Chad Preisler wrote: > >>>>> Just a quick update. Setting max.task.idle.ms to 10000 (10 seconds) > >>>> had no > >>>>> effect on this issue. > >>>>> > >>>>> On Tue, Nov 2, 2021 at 6:55 PM Chad Preisler < > chad.preis...@gmail.com> > >>>>> wrote: > >>>>> > >>>>>> No unfortunately it is not the case. The table record is written > >>>> about 20 > >>>>>> seconds before the stream record. I’ll crank up the time tomorrow > and > >>>> see > >>>>>> what happens. > >>>>>> > >>>>>> On Tue, Nov 2, 2021 at 6:24 PM Matthias J. Sax <mj...@apache.org> > >>>> wrote: > >>>>>> > >>>>>>> Hard to tell, but as it seems that you can reproduce the issue, it > >>>> might > >>>>>>> be worth a try to increase the idle time further. > >>>>>>> > >>>>>>> I guess one corner case for stream-table join that is not resolved > >>>> yet > >>>>>>> is when stream and table record have the same timestamp... For this > >>>>>>> case, the table record might not be processed first. > >>>>>>> > >>>>>>> Could you hit this case? > >>>>>>> > >>>>>>> > >>>>>>> -Matthias > >>>>>>> > >>>>>>> On 11/2/21 3:13 PM, Chad Preisler wrote: > >>>>>>>> Thank you for the information. We are using the Kafka 3.0 client > >>>>>>> library. > >>>>>>>> We are able to reliably reproduce this issue in our test > environment > >>>>>>> now. I > >>>>>>>> removed my timestamp extractor, and I set the max.task.idle.ms to > >>>>>>> 2000. I > >>>>>>>> also turned on trace logging for package > >>>>>>>> org.apache.kafka.streams.processor.internals. > >>>>>>>> > >>>>>>>> To create the issue we stopped the application and ran enough data > >>>> to > >>>>>>>> create a lag of 400 messages. We saw 5 missed joins. > >>>>>>>> > >>>>>>>> From the stream-thread log messages we saw the event message, > our > >>>>>>> stream > >>>>>>>> missed the join, and then several milliseconds later we saw the > >>>>>>>> stream-thread print out the status message. The stream-thread > >>>> printed > >>>>>>> out > >>>>>>>> our status message a total of 5 times. > >>>>>>>> > >>>>>>>> Given that only a few milliseconds passed between missing the join > >>>> and > >>>>>>> the > >>>>>>>> stream-thread printing the status message, would increasing the > >>>>>>>> max.task.idle.ms help? > >>>>>>>> > >>>>>>>> Thanks, > >>>>>>>> Chad > >>>>>>>> > >>>>>>>> On Mon, Nov 1, 2021 at 10:03 PM Matthias J. Sax <mj...@apache.org > > > >>>>>>> wrote: > >>>>>>>> > >>>>>>>>> Timestamp synchronization is not perfect, and as a matter of > fact, > >>>> we > >>>>>>>>> fixed a few gaps in 3.0.0 release. We actually hope, that we > >>>> closed the > >>>>>>>>> last gaps in 3.0.0... *fingers-crossed* :) > >>>>>>>>> > >>>>>>>>>> We are using a timestamp extractor that returns 0. > >>>>>>>>> > >>>>>>>>> You can do this, and it effectively "disables" timestamp > >>>>>>> synchronization > >>>>>>>>> as records on the KTable side don't have a timeline any longer. > As > >>>> a > >>>>>>>>> side effect it also allows you to "bootstrap" the table, as > records > >>>>>>> with > >>>>>>>>> timestamp zero will always be processed first (as they are > >>>> smaller). Of > >>>>>>>>> course, you also don't have time synchronization for "future" > data > >>>> and > >>>>>>>>> your program becomes non-deterministic if you reprocess old data. > >>>>>>>>> > >>>>>>>>>> his seemed to be the only > >>>>>>>>>> way to bootstrap enough records at startup to avoid the missed > >>>> join. > >>>>>>>>> > >>>>>>>>> Using 3.0.0 and enabling timestamp synchronization via > >>>>>>>>> `max.task.idle.ms` config, should allow you to get the correct > >>>>>>> behavior > >>>>>>>>> without the zero-extractor (of course, your KTable data must have > >>>>>>>>> smaller timestamps that your KStream data). > >>>>>>>>> > >>>>>>>>>> If I use "timestamp synchronization" do I have to remove the > zero > >>>>>>>>>> timestamp extractor? If I remove the zero timestamp extractor > will > >>>>>>>>>> timestamp synchronization take care of the missed join issue on > >>>>>>> startup? > >>>>>>>>> > >>>>>>>>> To be more precise: timestamp synchronization is _always_ on. The > >>>>>>>>> question is just how strict it is applied. By default, we do the > >>>>>>> weakest > >>>>>>>>> from which is only best effort. > >>>>>>>>> > >>>>>>>>>> I'm guessing the issue here is that occasionally the poll > request > >>>> is > >>>>>>> not > >>>>>>>>>> returning the matching record for the KTable side of the join > >>>> before > >>>>>>> the > >>>>>>>>>> task goes off and starts processing records. > >>>>>>>>> > >>>>>>>>> Yes, because of default best effort approach. That is why you > >>>> should > >>>>>>>>> increase `max.task.idle.ms` to detect this case and "skip" > >>>> processing > >>>>>>>>> and let KS do another poll() to get KTable data. > >>>>>>>>> > >>>>>>>>> 2.8 and earlier: > >>>>>>>>> > >>>>>>>>> max.task.idle.ms=0 -> best effort (no poll() retry) > >>>>>>>>> max.task.idle.ms>0 -> try to do another poll() until data is > >>>> there or > >>>>>>>>> idle time passed > >>>>>>>>> > >>>>>>>>> Note: >0 might still "fail" even if there is data, because > consumer > >>>>>>>>> fetch behavior is not predictable. > >>>>>>>>> > >>>>>>>>> > >>>>>>>>> 3.0: > >>>>>>>>> > >>>>>>>>> max.task.idle.ms=-1 -> best effort (no poll() retry) > >>>>>>>>> max.task.idle.ms=0 -> if there is data broker side, repeat to > >>>> poll() > >>>>>>>>> until you get the data > >>>>>>>>> max.task.idle.ms>0 -> even if there is not data broker side, > wait > >>>>>>> until > >>>>>>>>> data becomes available or the idle time passed > >>>>>>>>> > >>>>>>>>> > >>>>>>>>> Hope this helps. > >>>>>>>>> > >>>>>>>>> > >>>>>>>>> -Matthias > >>>>>>>>> > >>>>>>>>> On 11/1/21 4:29 PM, Guozhang Wang wrote: > >>>>>>>>>> Hello Chad, > >>>>>>>>>> > >>>>>>>>>> From your earlier comment, you mentioned "In my scenario the > >>>> records > >>>>>>>>> were > >>>>>>>>>> written to the KTable topic before the record was written to the > >>>>>>> KStream > >>>>>>>>>> topic." So I think Matthias and others have excluded this > >>>> possibility > >>>>>>>>> while > >>>>>>>>>> trying to help investigate. > >>>>>>>>>> > >>>>>>>>>> If only the matching records from KStream are returned via a > >>>> single a > >>>>>>>>>> consumer poll call but not the other records from KTable, then > you > >>>>>>> would > >>>>>>>>>> miss this matched join result. > >>>>>>>>>> > >>>>>>>>>> Guozhang > >>>>>>>>>> > >>>>>>>>>> > >>>>>>>>>> On Sun, Oct 31, 2021 at 7:28 AM Chad Preisler < > >>>>>>> chad.preis...@gmail.com> > >>>>>>>>>> wrote: > >>>>>>>>>> > >>>>>>>>>>> Thank you for your response and the links to the presentations. > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> *However, this seems tobe orthogonal to your issue?* > >>>>>>>>>>> > >>>>>>>>>>> Yes. From what I see in the code it looks like you have a > single > >>>>>>>>> consumer > >>>>>>>>>>> subscribed to multiple topics. Please correct me if I'm wrong. > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> *By default, timestamp synchronization is disabled. > >>>> Maybeenabling it > >>>>>>>>> would > >>>>>>>>>>> help?* > >>>>>>>>>>> > >>>>>>>>>>> We are using a timestamp extractor that returns 0. We did that > >>>>>>> because > >>>>>>>>> we > >>>>>>>>>>> were almost always missing joins on startup, and this seemed to > >>>> be > >>>>>>> the > >>>>>>>>> only > >>>>>>>>>>> way to bootstrap enough records at startup to avoid the missed > >>>> join. > >>>>>>> We > >>>>>>>>>>> found a post that said doing that would make the KTable act > like > >>>> the > >>>>>>>>>>> GlobalKTable at startup. So far this works great, we never > miss a > >>>>>>> join > >>>>>>>>> on a > >>>>>>>>>>> startup. If I use "timestamp synchronization" do I have to > >>>> remove the > >>>>>>>>> zero > >>>>>>>>>>> timestamp extractor? If I remove the zero timestamp extractor > >>>> will > >>>>>>>>>>> timestamp synchronization take care of the missed join issue on > >>>>>>> startup? > >>>>>>>>>>> > >>>>>>>>>>> I'm guessing the issue here is that occasionally the poll > >>>> request is > >>>>>>> not > >>>>>>>>>>> returning the matching record for the KTable side of the join > >>>> before > >>>>>>> the > >>>>>>>>>>> task goes off and starts processing records. Later when we put > >>>> the > >>>>>>> same > >>>>>>>>>>> record on the topic and the KTable has had a chance to load > more > >>>>>>> records > >>>>>>>>>>> the join works and everything is good to go. Because of the way > >>>> our > >>>>>>>>> system > >>>>>>>>>>> works no new status records have been written and so the new > >>>> record > >>>>>>>>> joins > >>>>>>>>>>> against the correct status. > >>>>>>>>>>> > >>>>>>>>>>> Do you agree that the poll request is returning the KStream > >>>> record > >>>>>>> but > >>>>>>>>> not > >>>>>>>>>>> returning the KTable record and therefore the join is getting > >>>>>>> missed? If > >>>>>>>>>>> you don't agree, what do you think is going on? Is there a way > to > >>>>>>> prove > >>>>>>>>>>> this out? > >>>>>>>>>>> > >>>>>>>>>>> Thanks, > >>>>>>>>>>> Chad > >>>>>>>>>>> > >>>>>>>>>>> On Sat, Oct 30, 2021 at 2:09 PM Matthias J. Sax < > >>>> mj...@apache.org> > >>>>>>>>> wrote: > >>>>>>>>>>> > >>>>>>>>>>>> Yes, a StreamThread has one consumer. The number of > >>>> StreamThreads > >>>>>>> per > >>>>>>>>>>>> instance is configurable via `num.stream.threads`. Partitions > >>>> are > >>>>>>>>>>>> assigned to threads similar to consumer is a plain consumer > >>>> group. > >>>>>>>>>>>> > >>>>>>>>>>>> It seems you run with the default of one thread per instance. > >>>> As you > >>>>>>>>>>>> spin up 12 instances, it results in 12 threads for the > >>>> application. > >>>>>>> As > >>>>>>>>>>>> you have 12 partitions, using more threads won't be useful as > no > >>>>>>>>>>>> partitions are left for them to process. > >>>>>>>>>>>> > >>>>>>>>>>>> For a stream-table joins, there will be one task per > "partition > >>>>>>> pair" > >>>>>>>>>>>> that computes the join for those partitions. So you get 12 > >>>> tasks, > >>>>>>> and > >>>>>>>>>>>> each thread processes one task in your setup. Ie, a thread > >>>> consumer > >>>>>>> is > >>>>>>>>>>>> reading data for both input topics. > >>>>>>>>>>>> > >>>>>>>>>>>> Pausing happens on a per-partition bases: for joins there is > two > >>>>>>>>> buffers > >>>>>>>>>>>> per task (one for each input topic partition). It's possible > >>>> that > >>>>>>> one > >>>>>>>>>>>> partition is paused while the other is processed. However, > this > >>>>>>> seems > >>>>>>>>> to > >>>>>>>>>>>> be orthogonal to your issue? > >>>>>>>>>>>> > >>>>>>>>>>>> For a GlobalKTable, you get an additional GlobalThread that > only > >>>>>>> reads > >>>>>>>>>>>> the data from the "global topic" to update the GlobalKTable. > >>>>>>> Semantics > >>>>>>>>>>>> of KStream-KTable and KStream-GlobalKTable joins are > different: > >>>> Cf > >>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>> > >>>>>>> > >>>> > https://www.confluent.io/events/kafka-summit-europe-2021/temporal-joins-in-kafka-streams-and-ksqldb/ > >>>>>>>>>>>> > >>>>>>>>>>>> For the timestamp synchronization, you may checkout ` > >>>>>>> max.task.idle.ms` > >>>>>>>>>>>> config. By default, timestamp synchronization is disabled. > Maybe > >>>>>>>>>>>> enabling it would help? > >>>>>>>>>>>> > >>>>>>>>>>>> You may also check out slides 34-38: > >>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>> > >>>>>>> > >>>> > https://www.confluent.io/kafka-summit-san-francisco-2019/whats-the-time-and-why/ > >>>>>>>>>>>> > >>>>>>>>>>>> There is one corner case: if two records with the same > timestamp > >>>>>>> come > >>>>>>>>>>>> it, it's not defined which one will be processed first. > >>>>>>>>>>>> > >>>>>>>>>>>> Hope this helps. > >>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>>>>> -Matthias > >>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>>>>> On 10/30/21 6:45 AM, Chad Preisler wrote: > >>>>>>>>>>>>> Yes, this helped. I have some additional questions. > >>>>>>>>>>>>> > >>>>>>>>>>>>> Does StreamThread have one consumer? (Looks like it, but just > >>>> want > >>>>>>> to > >>>>>>>>>>>>> confirm) > >>>>>>>>>>>>> Is there a separate StreamThread for each topic including the > >>>>>>> KTable? > >>>>>>>>>>>>> If a KTable is a StreamThread and there is a StreamTask for > >>>> that > >>>>>>>>>>> KTable, > >>>>>>>>>>>>> could my buffer be getting filled up, and the mainConsumer > for > >>>> the > >>>>>>>>>>> KTable > >>>>>>>>>>>>> be getting paused? I see this code in StreamTask#addRecords. > >>>>>>>>>>>>> > >>>>>>>>>>>>> // if after adding these records, its partition queue's > >>>> buffered > >>>>>>> size > >>>>>>>>>>> has > >>>>>>>>>>>>> been > >>>>>>>>>>>>> // increased beyond the threshold, we can then > >>>> pause the > >>>>>>>>>>>>> consumption for this partition > >>>>>>>>>>>>> if (newQueueSize > maxBufferedSize) { > >>>>>>>>>>>>> mainConsumer.pause(singleton(partition)); > >>>>>>>>>>>>> } > >>>>>>>>>>>>> > >>>>>>>>>>>>> Is there any specific logging that I can set to debug or > trace > >>>> that > >>>>>>>>>>> would > >>>>>>>>>>>>> help me troubleshoot? I'd prefer not to turn debug and/or > >>>> trace on > >>>>>>> for > >>>>>>>>>>>>> every single class. > >>>>>>>>>>>>> > >>>>>>>>>>>>> Thanks, > >>>>>>>>>>>>> Chad > >>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>>>> On Sat, Oct 30, 2021 at 5:20 AM Luke Chen <show...@gmail.com > > > >>>>>>> wrote: > >>>>>>>>>>>>> > >>>>>>>>>>>>>> Hi Chad, > >>>>>>>>>>>>>>> I'm wondering if someone can point me to the Kafka streams > >>>>>>> internal > >>>>>>>>>>>> code > >>>>>>>>>>>>>> that reads records for the join? > >>>>>>>>>>>>>> --> You can check StreamThread#pollPhase, where stream > thread > >>>>>>> (main > >>>>>>>>>>>>>> consumer) periodically poll records. And then, it'll process > >>>> each > >>>>>>>>>>>> topology > >>>>>>>>>>>>>> node with these polled records in stream tasks > >>>>>>> (StreamTask#process). > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> Hope that helps. > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> Thanks. > >>>>>>>>>>>>>> Luke > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> On Sat, Oct 30, 2021 at 5:42 PM Gilles Philippart > >>>>>>>>>>>>>> <gilles.philipp...@fundingcircle.com.invalid> wrote: > >>>>>>>>>>>>>> > >>>>>>>>>>>>>>> Hi Chad, this talk around 24:00 clearly explains what > you’re > >>>>>>> seeing > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>> > >>>>>>> > >>>> > https://www.confluent.io/events/kafka-summit-europe-2021/failing-to-cross-the-streams-lessons-learned-the-hard-way/ > >>>>>>>>>>>>>>> < > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>> > >>>>>>> > >>>> > https://www.confluent.io/events/kafka-summit-europe-2021/failing-to-cross-the-streams-lessons-learned-the-hard-way/ > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> Gilles > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> On 30 Oct 2021, at 04:02, Chad Preisler < > >>>>>>> chad.preis...@gmail.com> > >>>>>>>>>>>>>> wrote: > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> Hello, > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> I have a stream application that does a KStream to KTable > >>>> left > >>>>>>>>> join. > >>>>>>>>>>>> We > >>>>>>>>>>>>>>>> seem to be occasionally missing joins > >>>>>>> < > >>>> > https://www.google.com/maps/search/%3E%3E%3E%3E%3E+seem+to+be+occasionally+missing+joins?entry=gmail&source=g > >>>>> > >>>>>>> (KTable side is null). > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> I'm wondering if someone can point me to the Kafka streams > >>>>>>> internal > >>>>>>>>>>>>>> code > >>>>>>>>>>>>>>>> that reads records for the join? I've poked around the > Kafka > >>>>>>> code > >>>>>>>>>>>> base, > >>>>>>>>>>>>>>> but > >>>>>>>>>>>>>>>> there is a lot there. I imagine there is some consumer > poll > >>>> for > >>>>>>>>> each > >>>>>>>>>>>>>> side > >>>>>>>>>>>>>>>> of the join, and possibly a background thread for reading > >>>> the > >>>>>>>>> KTable > >>>>>>>>>>>>>>> topic. > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> I figure there are several possible causes of this issue, > >>>> and > >>>>>>> since > >>>>>>>>>>>>>>> nothing > >>>>>>>>>>>>>>>> is really jumping out in my code, I was going to start > >>>> looking > >>>>>>> at > >>>>>>>>>>> the > >>>>>>>>>>>>>>> Kafka > >>>>>>>>>>>>>>>> code to see if there is something I can do to fix this. > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> Thanks, > >>>>>>>>>>>>>>>> Chad > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> -- > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> Funding Circle Limited is authorised and regulated by the > >>>>>>> Financial > >>>>>>>>>>>>>>> Conduct Authority under firm registration number 722513. > >>>> Funding > >>>>>>>>>>> Circle > >>>>>>>>>>>>>> is > >>>>>>>>>>>>>>> not covered by the Financial Services Compensation Scheme. > >>>>>>>>> Registered > >>>>>>>>>>>> in > >>>>>>>>>>>>>>> England (Co. No. 06968588) with registered office at 71 > Queen > >>>>>>>>>>> Victoria > >>>>>>>>>>>>>>> Street, London EC4V 4AY. > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>> > >>>>>>>>>> > >>>>>>>>> > >>>>>>>> > >>>>>>> > >>>>>> > >>>>> > >>>> > >>> > > >
