pvillard31 commented on code in PR #11702:
URL: https://github.com/apache/nifi/pull/11702#discussion_r4070587283
##########
nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/OffsetTracker.java:
##########
@@ -24,25 +24,34 @@
import java.util.HashMap;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.atomic.AtomicLong;
public class OffsetTracker {
private final Map<TopicPartitionSummary, OffsetSummary> offsets = new
HashMap<>();
private final Map<String, Long> recordCounts = new HashMap<>();
+ private final Map<TopicPartitionSummary, Long> partitionRecords = new
HashMap<>();
+ private final Map<TopicPartitionSummary, Long> partitionBytes = new
HashMap<>();
private final AtomicLong totalRecordSize = new AtomicLong();
public void update(final ByteRecord consumerRecord) {
final TopicPartitionSummary topicPartitionSummary = new
TopicPartitionSummary(consumerRecord.getTopic(), consumerRecord.getPartition());
final long offset = consumerRecord.getOffset();
final OffsetSummary offsetSummary =
offsets.computeIfAbsent(topicPartitionSummary, (summary) -> new
OffsetSummary(offset));
offsetSummary.setOffset(offset);
- recordCounts.merge(consumerRecord.getTopic(),
consumerRecord.getBundledCount(), Long::sum);
- // Update Total Record Size with Key and Value length
- consumerRecord.getKey()
- .map(key -> key.length)
- .ifPresent(totalRecordSize::addAndGet);
- totalRecordSize.addAndGet(consumerRecord.getValue().length);
+ final long bundledCount = consumerRecord.getBundledCount();
+ recordCounts.merge(consumerRecord.getTopic(), bundledCount, Long::sum);
+ partitionRecords.merge(topicPartitionSummary, bundledCount, Long::sum);
+
+ long recordSize = consumerRecord.getValue().length;
Review Comment:
Should `kafka.bytes.consumed` be measured from the original Kafka records
before DEMARCATOR bundling, since the current path counts added demarcator
bytes and can omit original keys?
##########
nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaTest.java:
##########
@@ -231,6 +257,81 @@ public void testOnStoppedPreservesLastCaughtUp() throws
Exception {
assertEquals(OptionalLong.of(5L), afterStop.get().getRecordCount());
}
+ @Test
Review Comment:
Can we add coverage for `kafka.bytes.consumed` with DEMARCATOR enabled,
including multiple keyed records, to verify that the metric reports the
original Kafka bytes?
--
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]
