exceptionfactory commented on code in PR #11702:
URL: https://github.com/apache/nifi/pull/11702#discussion_r4072737531
##########
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:
Yes, that sounds like the right approach, I will take a closer look at
tracking the original record 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]
