kennknowles commented on PR #25554: URL: https://github.com/apache/beam/pull/25554#issuecomment-2588473747
> > If I understand that correctly (and I'm definitely not sure about that :)), then you refer to the time when metrics are compute in the source (KafkaSource in this case). But the wrapper should update the metric container on each call to `advance`, see [1]. I wonder why it is needed to do the same when doing checkpoint (though it sort might make more sense than to do that on each call to `advance()`).- I'm not saying it is wrong, I'd just like to understand the reason. > > Backlog is reported from getCheckpointMark(), which is done by some other thread. Not sure why it is done there. But this is the main issue. Advance function runs on main thread and it has Metric context so I am able to see element count metrics. However Checkpoint thread does not have context thats why It can not emit metrics. > > If I move reportBacklog() function in advance function i am able to see backlog metrics too. We could do that in advance(), but that would unnecessary overhead for every single record. :) In current code `advance()` calls `nextBatch()` which calls `reportBacklog()`: https://github.com/apache/beam/blob/4611a81b224757234d027661927f9725253b0fba/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java#L674 So there might be some other issue. But if there is a call to `reportBacklog()` in `getCheckpointMark` in KafkaIO it is a good thing for the runner to provide a working metrics container. I don't understand this area of the FlinkRunner well enough to know why each method is custom with a temporary container, but we should do it. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
