You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by yh...@apache.org on 2024/05/14 16:30:10 UTC
(beam) branch master updated: Set backlog in gauge metric (#31137)
This is an automated email from the ASF dual-hosted git repository.
yhu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new ee170e142b9 Set backlog in gauge metric (#31137)
ee170e142b9 is described below
commit ee170e142b96401ef83c0e06e8080281a42b2f77
Author: Naireen Hussain <na...@live.com>
AuthorDate: Tue May 14 09:30:03 2024 -0700
Set backlog in gauge metric (#31137)
Co-authored-by: Naireen <na...@google.com>
---
.../apache/beam/sdk/io/kafka/KafkaUnboundedReader.java | 18 +++++++++++++++---
1 file changed, 15 insertions(+), 3 deletions(-)
diff --git a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java
index 054eb502cd8..fed03047cf1 100644
--- a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java
+++ b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java
@@ -23,6 +23,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
+import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -224,6 +225,9 @@ class KafkaUnboundedReader<K, V> extends UnboundedReader<KafkaRecord<K, V>> {
METRIC_NAMESPACE, RAW_SIZE_METRIC_PREFIX + pState.topicPartition.toString());
rawSizes.update(recordSize);
+ for (Map.Entry<String, Long> backlogSplit : perPartitionBacklogMetrics.entrySet()) {
+ backlogBytesOfSplit.set(backlogSplit.getValue());
+ }
return true;
} else { // -- (b)
@@ -341,6 +345,7 @@ class KafkaUnboundedReader<K, V> extends UnboundedReader<KafkaRecord<K, V>> {
private final Counter bytesReadBySplit;
private final Gauge backlogBytesOfSplit;
private final Gauge backlogElementsOfSplit;
+ private HashMap<String, Long> perPartitionBacklogMetrics = new HashMap<String, Long>();;
private final Counter checkpointMarkCommitsEnqueued =
Metrics.counter(METRIC_NAMESPACE, CHECKPOINT_MARK_COMMITS_ENQUEUED_METRIC);
// Checkpoint marks skipped in favor of newer mark (only the latest needs to be committed).
@@ -491,6 +496,10 @@ class KafkaUnboundedReader<K, V> extends UnboundedReader<KafkaRecord<K, V>> {
lastWatermark = timestampPolicy.getWatermark(mkTimestampPolicyContext());
return lastWatermark;
}
+
+ String name() {
+ return this.topicPartition.toString();
+ }
}
KafkaUnboundedReader(
@@ -528,14 +537,16 @@ class KafkaUnboundedReader<K, V> extends UnboundedReader<KafkaRecord<K, V>> {
prevWatermark = Optional.of(new Instant(ckptMark.getWatermarkMillis()));
}
- states.add(
- new PartitionState<>(
+ PartitionState<K, V> state =
+ new PartitionState<K, V>(
tp,
nextOffset,
source
.getSpec()
.getTimestampPolicyFactory()
- .createTimestampPolicy(tp, prevWatermark)));
+ .createTimestampPolicy(tp, prevWatermark));
+ states.add(state);
+ perPartitionBacklogMetrics.put(state.name(), 0L);
}
partitionStates = ImmutableList.copyOf(states);
@@ -717,6 +728,7 @@ class KafkaUnboundedReader<K, V> extends UnboundedReader<KafkaRecord<K, V>> {
if (pBacklog == UnboundedReader.BACKLOG_UNKNOWN) {
return UnboundedReader.BACKLOG_UNKNOWN;
}
+ perPartitionBacklogMetrics.put(p.name(), pBacklog);
backlogCount += pBacklog;
}