Thanks for the follow up Boyang and Guozhang, I have updated the kip to include these ideas.
Guozhang, that is a good idea about using the TaskMetadata. We can get it through the ThreadMetadata with a minor change to `localThreadMetadata` in kafkaStreams. This means that we will only need to update TaskMetadata and add no other APIs Boyang, since each TaskMetadata contains the TaskId and TopicPartitions I don't believe mapping either way will be a problem. Also I think we can do something like record the time the task started idling and when it stops idling we can override it to -1. I think that should clear up the first two points. As for your third point I am not sure I 100% understand. The ThreadMetadata will contain a set of all task assigned to that thread. Any health check service will just need to query all clients and aggregate their responses to get a complete picture of all tasks correct? walker On Thu, Feb 25, 2021 at 9:57 AM Guozhang Wang <wangg...@gmail.com> wrote: > Regarding the second API and the `TaskStatus` class: I'd suggest we > consolidate on the existing `TaskMetadata` since we have already > accumulated a bunch of such classes, and its better to keep them small as > public APIs. You can see https://issues.apache.org/jira/browse/KAFKA-12370 > for a reference and a proposal. > > On Thu, Feb 25, 2021 at 9:40 AM Boyang Chen <reluctanthero...@gmail.com> > wrote: > > > Thanks for the updates Walker. Some replies and follow-up questions: > > > > 1. I agree one task could have multiple partitions, but when we hit a > delay > > in terms of offset progress, do we have a convenient way to reverse > mapping > > TopicPartition to the problematic task? In production, I believe it would > > be much quicker to identify the problem using task.id instead of topic > > partition, especially when it points to an internal topic. I think having > > the task id as part of the entry value seems useful, which means getting > > something like Map<TopicPartition, TaskProgress> where TaskProgress > > contains both committed offsets & task id. > > > > 2. The task idling API was still confusing. I don't think we care about > the > > exact state when making tasksIdling()query, instead we care more about > how > > long one task has been in idle state since when you called, which > reflects > > whether it is a normal idling period. So I feel it might be helpful to > > track that time difference and report it in the TaskStatus struct. > > > > 3. What I want to achieve to have some global mapping of either > > TopicPartition or TaskId was that it is not possible for a health check > > service to report a task failure that doesn't emit any metrics. So as > long > > as we have a global topic partition API, health check could always be > aware > > of any task/partition not reporting its progress, does that make sense? > If > > you feel we have a better way to achieve this, such as querying all the > > input/intermediate topic metadata directly from Kafka for the baseline, I > > think that should be good as well and worth mentioning it in the KIP. > > > > Also it seems that the KIP hasn't reflected what you proposed for the > task > > idling status. > > > > Best, > > Boyang > > > > > > On Wed, Feb 24, 2021 at 9:11 AM Walker Carlson <wcarl...@confluent.io> > > wrote: > > > > > Thank you for the comments everyone! > > > > > > I think there are a few things I can clear up in general then I will > > > specifically respond to each question. > > > > > > First, when I say "idling" I refer to task idling. Where the stream is > > > intentionally not making progress. ( > > > https://issues.apache.org/jira/browse/KAFKA-10091 is an example). This > > > becomes relevant if a task is waiting on one partition with no data but > > > that is holding up a partition with data. That would cause one just > > looking > > > at the committed offset changes to believe the task has a problem when > it > > > is working as intended. > > > > > > In light of this confusion. I plan to change tasksIdling() to > > `Map<TaskId, > > > TaskStatus> getTasksStatus()` this should hopefully make it more clear > > what > > > is being exposed. > > > > > > TaskStatus would include: TopicPartions, TaskId, ProcessorTopology, > > Idling, > > > and State. > > > > > > Boyang: > > > > > > 2) I think that each task should report on whatever TopicPartitions > they > > > hold, this means a Topic Partition might get reported twice but the > user > > > can roll those up and use the larger one when looking at the whole app. > > > > > > 4) If the user collects the committed offsets across all the running > > > clients there shouldn't be any tasks missing correct? > > > > > > 6) Because there is not a 1:1 mapping between Tasks and > TopicPartitions I > > > think it is cleaner to report them separately. > > > > > > Guozhang: > > > > > > 1) Yes, that was my original plan but it made more sense to mirror how > > the > > > consumer exposes the committed offset. > > > > > > 3) That is a good point. I think that we should include internal topics > > as > > > well. I think that if the topology were to evolve there should be fair > > > warning anyways. Maybe you can clarify what would be limited by > exposing > > > the interior topics here? I thought a user could find them in other > ways. > > > If it is the name we could aynomise them before exposing them. > > > > > > Thank you all for your comments. If I did not respond directly to one > of > > > your questions I updated the kip to include the details it was > > requesting. > > > I didn't not include my proposed changes mentioned earlier as I would > > like > > > to get some feedback about what to include in TaskStatus and in > general. > > > > > > best, > > > Walker > > > > > > On Mon, Feb 22, 2021 at 10:20 PM Guozhang Wang <wangg...@gmail.com> > > wrote: > > > > > > > Hello Walker, thanks for the KIP. A few thoughts: > > > > > > > > 1) Have you considered just relying on the `KafkaStreams#metrics()` > > that > > > > includes embedded consumer metrics that have the committed offsets > > > > instead of adding a new API? Not advocating that this is a better > > > approach > > > > but want to make sure we considered all options before we come to the > > > "last > > > > resort" of adding new public interfaces. > > > > > > > > 2) The javadoc mentions "tasks assigned to this client", but the > > returned > > > > map is on partitions. I think we should make the javadoc and the > return > > > > types consistent, either tasks or topic partitions. > > > > > > > > 3) In addition, if for 2) above we ended up with topic partitions, > then > > > > would they include only external source topics, or also including > > > internal > > > > repartition / changelog topics? I think including only external > source > > > > topic partitions are not sufficient for your goal of tracking > progress, > > > but > > > > exposing internal topic names are also a big commitment here for > future > > > > topology evolution. > > > > > > > > 4) For "tasksIdling", I'm wondering if we can make it more general, > > that > > > > the returned value is not just a boolean, but a TaskState that can be > > an > > > > enum of "created, restoring, running, idle, closing". This could help > > us > > > in > > > > the future to track other things like restoration efficiency and > > > rebalance > > > > efficiency etc. > > > > > > > > 5) We need to clarify how is "idling" being defined here: e.g. we can > > > > clearly state that a task is considered idle only if 1) lag is > > > > increasing, indicating that there are indeed new records arrived at > > > source, > > > > while committed offset is not advancing, AND 2) produced offset > > (imagine > > > we > > > > may have punctuations that generate new data to the output topic even > > if > > > > there's no input for a while) is not advancing either. > > > > > > > > > > > > Guozhang > > > > > > > > > > > > > > > > On Mon, Feb 22, 2021 at 3:11 PM Boyang Chen < > > reluctanthero...@gmail.com> > > > > wrote: > > > > > > > > > Thanks Walker for the proposed KIP! This should definitely empower > > > > KStream > > > > > users with better visibility. > > > > > > > > > > Meanwhile I got a couple of questions/suggestions: > > > > > > > > > > > > > > > 1. typo "repost/report" in the motivation section. > > > > > > > > > > 2. What offsets do we report when the task is under restoration or > > > > > rebalancing? > > > > > > > > > > 3. IIUC, we should clearly state that our reported metrics are > based > > > off > > > > > locally assigned tasks for each instance. > > > > > > > > > > 4. In the meantime, what’s our strategy to report tasks that are > not > > > > local > > > > > to the instance? Users would normally try to monitor all the > possible > > > > > tasks, and it’s unfortunate we couldn’t determine whether we have > > lost > > > > > tasks. My brainstorming was whether it makes sense for the leader > > > > instance > > > > > to report the task progress as -1 for all “supposed to be running” > > > tasks, > > > > > so that on the metrics collector side it could catch any missing > > tasks. > > > > > > > > > > 5. It seems not clear how users should use `isTaskIdling`. Why not > > > > report a > > > > > map/set for idling tasks just as what we did for committed offsets? > > > > > > > > > > 6. Why do we use TopicPartition instead of TaskId as the key in the > > > > > returned map? > > > > > 7. Could we include some details in where we got the commit offsets > > for > > > > > each task? Is it through consumer offset fetch, or the stream > > > processing > > > > > progress based on the records fetched? > > > > > > > > > > > > > > > On Mon, Feb 22, 2021 at 3:00 PM Walker Carlson < > > wcarl...@confluent.io> > > > > > wrote: > > > > > > > > > > > Hello all, > > > > > > > > > > > > I would like to start discussion on KIP-715. This kip aims to > make > > it > > > > > > easier to monitor Kafka Streams progress by exposing the > committed > > > > offset > > > > > > in a similar way as the consumer client does. > > > > > > > > > > > > Here is the KIP: https://cwiki.apache.org/confluence/x/aRRRCg > > > > > > > > > > > > Best, > > > > > > Walker > > > > > > > > > > > > > > > > > > > > > > > -- > > > > -- Guozhang > > > > > > > > > > > > -- > -- Guozhang >
