Naireen commented on code in PR #33503:
URL: https://github.com/apache/beam/pull/33503#discussion_r1979999342
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java:
##########
@@ -146,14 +154,30 @@ private void recordRpcLatencyMetrics() {
}
}
- private void recordBacklogBytes() {
+ private void recordBacklogBytesInternal() {
for (Map.Entry<String, Long> backlogs :
perTopicPartitionBacklogs().entrySet()) {
Gauge gauge =
KafkaSinkMetrics.createBacklogGauge(MetricName.named("KafkaSink",
backlogs.getKey()));
gauge.set(backlogs.getValue());
}
}
+ /**
+ * This is for recording backlog bytes on the current thread.
+ *
+ * @param topicName topicName
+ * @param partitionId partitionId for the topic Only included in the
metric key if
+ * 'supportsMetricsDeletion' is enabled.
+ * @param backlogBytes backlog for the topic Only included in the metric
key if
+ * 'supportsMetricsDeletion' is enabled.
+ */
+ @Override
+ public void recordBacklogBytes(String topicName, int partitionId, long
backlogBytes) {
Review Comment:
Added comments, one is for updating the metrics container on the current
thread, and one is for when its not on the current thread, so you update a map
and record values, and once you're on the main processing thread of the dofn,
then you update/create gauges with all the observed values (this is from the
legacy implementation since metrics are updated in the background).
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java:
##########
@@ -39,6 +40,8 @@ public interface KafkaMetrics {
void updateKafkaMetrics();
+ void recordBacklogBytes(String topic, int partitionId, long backlog);
Review Comment:
We could, where we store all values in a map, and then create gauges
afterwards, but we don't need to in this case, we can directly create the
gauges. See comments below
--
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]
