This is an automated email from the ASF dual-hosted git repository.
pvillard31 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 856e5fd0a9d NIFI-16375 Add Lag and Record Metrics to ConsumeKafka
(#11702)
856e5fd0a9d is described below
commit 856e5fd0a9da89d3efdc2d2cab447c7d9d23df5c
Author: David Handermann <[email protected]>
AuthorDate: Wed Sep 23 03:25:45 2026 -0500
NIFI-16375 Add Lag and Record Metrics to ConsumeKafka (#11702)
---
.../nifi/kafka/processors/ConsumeKafkaIT.java | 9 ++
.../apache/nifi/kafka/processors/ConsumeKafka.java | 103 ++++++++++++++-
.../kafka/processors/consumer/KafkaMetricName.java | 37 ++++++
.../kafka/processors/consumer/OffsetTracker.java | 34 ++++-
.../AbstractRecordStreamKafkaMessageConverter.java | 8 ++
.../nifi/kafka/processors/ConsumeKafkaTest.java | 147 +++++++++++++++++++++
.../processors/consumer/OffsetTrackerTest.java | 131 ++++++++++++++++++
.../RecordStreamKafkaMessageConverterTest.java | 65 +++++++--
8 files changed, 517 insertions(+), 17 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaIT.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaIT.java
index 76fe12d06ad..50daa420b83 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaIT.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaIT.java
@@ -20,6 +20,7 @@ import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.nifi.controller.ControllerService;
+import org.apache.nifi.kafka.processors.consumer.KafkaMetricName;
import org.apache.nifi.kafka.processors.consumer.ProcessingStrategy;
import org.apache.nifi.kafka.service.Kafka3ConnectionService;
import org.apache.nifi.kafka.service.api.consumer.AutoOffsetReset;
@@ -39,6 +40,7 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
+import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
@@ -115,6 +117,13 @@ class ConsumeKafkaIT extends AbstractConsumeKafkaIT {
flowFile.assertAttributeEquals("bbb", "value");
flowFile.assertAttributeNotExists("aaa");
flowFile.assertAttributeNotExists("ccc");
+
+ final Map<String, String> topicPartitionAttributes = Map.of(
+ KafkaFlowFileAttribute.KAFKA_TOPIC, topic,
+ KafkaFlowFileAttribute.KAFKA_PARTITION,
Integer.toString(FIRST_PARTITION));
+ assertEquals(1L,
runner.getCounterValue(KafkaMetricName.RECORDS_CONSUMED.getMetricName(),
topicPartitionAttributes));
+ assertEquals(RECORD_VALUE.getBytes(StandardCharsets.UTF_8).length,
+
runner.getCounterValue(KafkaMetricName.BYTES_CONSUMED.getMetricName(),
topicPartitionAttributes));
}
@Test
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/ConsumeKafka.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/ConsumeKafka.java
index 483f5466ff2..f000ea8b8be 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/ConsumeKafka.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/ConsumeKafka.java
@@ -36,6 +36,7 @@ import
org.apache.nifi.components.connector.components.ConnectorMethod;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.kafka.processors.common.HeaderValueConverter;
import org.apache.nifi.kafka.processors.common.KafkaUtils;
+import org.apache.nifi.kafka.processors.consumer.KafkaMetricName;
import org.apache.nifi.kafka.processors.consumer.OffsetTracker;
import org.apache.nifi.kafka.processors.consumer.ProcessingStrategy;
import org.apache.nifi.kafka.processors.consumer.bundle.ByteRecordBundler;
@@ -49,6 +50,7 @@ import
org.apache.nifi.kafka.processors.consumer.convert.RecordStreamKafkaMessag
import
org.apache.nifi.kafka.processors.consumer.convert.WrapperRecordStreamKafkaMessageConverter;
import org.apache.nifi.kafka.service.api.KafkaConnectionService;
import org.apache.nifi.kafka.service.api.common.PartitionState;
+import org.apache.nifi.kafka.service.api.common.TopicPartitionSummary;
import org.apache.nifi.kafka.service.api.consumer.AutoOffsetReset;
import org.apache.nifi.kafka.service.api.consumer.KafkaConsumerService;
import org.apache.nifi.kafka.service.api.consumer.PollingContext;
@@ -70,6 +72,7 @@ import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.VerifiableProcessor;
import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.metrics.CommitTiming;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.serialization.RecordReader;
import org.apache.nifi.serialization.RecordReaderFactory;
@@ -85,11 +88,13 @@ import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.HashSet;
import java.util.HexFormat;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.OptionalLong;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
@@ -500,7 +505,9 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
break;
}
- final Iterator<ByteRecord> consumerRecords =
consumerService.poll(maxWaitDuration).iterator();
+ final TopicPartitionScanningIterator consumerRecords = new
TopicPartitionScanningIterator(
+ consumerService.poll(maxWaitDuration).iterator(),
offsetTracker
+ );
if (!consumerRecords.hasNext()) {
getLogger().trace("No Kafka Records consumed: {}",
pollingContext);
// Check if a rebalance occurred during poll - if so,
break to commit what we have
@@ -513,6 +520,7 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
recordsReceived = true;
processConsumerRecords(context, session, offsetTracker,
consumerRecords);
+ recordLag(session, consumerService,
consumerRecords.getTopicPartitionSummaries());
// Check if a rebalance occurred during poll - if so,
break to commit what we have
if (consumerService.hasRevokedPartitions()) {
@@ -529,6 +537,8 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
}
} catch (final Exception e) {
getLogger().error("Failed to consume Kafka Records", e);
+ // Metrics recorded based on Processor behavior regardless
of exceptions
+ recordConsumption(session, offsetTracker);
consumerService.rollback();
close(consumerService, "Encountered Exception while
consuming or writing out Kafka Records");
context.yield();
@@ -539,6 +549,10 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
}
}
+ if (recordsReceived) {
+ recordConsumption(session, offsetTracker);
+ }
+
if (!recordsReceived && !consumerService.hasRevokedPartitions()) {
getLogger().trace("No Kafka Records consumed, re-queuing
consumer");
consumerServices.offer(consumerService);
@@ -572,6 +586,39 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
}
}
+ // Lag recorded for polled partitions that return records
+ private void recordLag(final ProcessSession session, final
KafkaConsumerService consumerService,
+ final Collection<TopicPartitionSummary> topicPartitions) {
+ for (final TopicPartitionSummary topicPartition : topicPartitions) {
+ final OptionalLong lag =
consumerService.currentLag(topicPartition);
+ if (lag.isPresent()) {
+
session.recordGauge(KafkaMetricName.RECORDS_CONSUMER_LAG.getMetricName(),
lag.getAsLong(),
+ topicPartitionAttributes(topicPartition.getTopic(),
topicPartition.getPartition()), CommitTiming.NOW);
+ }
+ }
+ }
+
+ private void recordConsumption(final ProcessSession session, final
OffsetTracker offsetTracker) {
+ for (final Map.Entry<TopicPartitionSummary, Long> entry :
offsetTracker.getPartitionRecords().entrySet()) {
+ final TopicPartitionSummary topicPartition = entry.getKey();
+
session.adjustCounter(KafkaMetricName.RECORDS_CONSUMED.getMetricName(),
entry.getValue(),
+ topicPartitionAttributes(topicPartition.getTopic(),
topicPartition.getPartition()), CommitTiming.NOW);
+ }
+
+ for (final Map.Entry<TopicPartitionSummary, Long> entry :
offsetTracker.getPartitionBytes().entrySet()) {
+ final TopicPartitionSummary topicPartition = entry.getKey();
+
session.adjustCounter(KafkaMetricName.BYTES_CONSUMED.getMetricName(),
entry.getValue(),
+ topicPartitionAttributes(topicPartition.getTopic(),
topicPartition.getPartition()), CommitTiming.NOW);
+ }
+ }
+
+ private static Map<String, String> topicPartitionAttributes(final String
topic, final int partition) {
+ return Map.of(
+ KafkaFlowFileAttribute.KAFKA_TOPIC, topic,
+ KafkaFlowFileAttribute.KAFKA_PARTITION,
Integer.toString(partition)
+ );
+ }
+
/**
* Reports backlog by querying committed and end offsets for the
Processor's consumer group via the
* connection service's Admin client. The Admin-client path is used
regardless of whether any
@@ -841,6 +888,7 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
getLogger().debug("Session committed successfully during
rebalance callback");
if (offsetTracker != null) {
+ recordConsumption(session, offsetTracker);
offsetTracker.getRecordCounts().forEach((topic, count)
-> {
session.adjustCounter("Records Acknowledged for "
+ topic, count, true);
});
@@ -950,6 +998,59 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
return pollingContext;
}
+ static class TopicPartitionScanningIterator implements
Iterator<ByteRecord> {
+ private final Iterator<ByteRecord> iterator;
+ private final OffsetTracker offsetTracker;
+ private final Set<TopicPartitionSummary> topicPartitionSummaries = new
HashSet<>();
+ private TopicPartitionSummary lastTopicPartition;
+
+ TopicPartitionScanningIterator(final Iterator<ByteRecord> iterator,
final OffsetTracker offsetTracker) {
+ this.iterator = iterator;
+ this.offsetTracker = offsetTracker;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return iterator.hasNext();
+ }
+
+ @Override
+ public ByteRecord next() {
+ final ByteRecord record = iterator.next();
+ final TopicPartitionSummary topicPartition =
topicPartition(record);
+ offsetTracker.recordConsumed(topicPartition,
record.getBundledCount(), consumedBytes(record));
+ return record;
+ }
+
+ public Set<TopicPartitionSummary> getTopicPartitionSummaries() {
+ return topicPartitionSummaries;
+ }
+
+ private TopicPartitionSummary topicPartition(final ByteRecord record) {
+ if (isNewTopicPartition(record)) {
+ final TopicPartitionSummary summary = new
TopicPartitionSummary(record.getTopic(), record.getPartition());
+ topicPartitionSummaries.add(summary);
+ lastTopicPartition = summary;
+ }
+ return lastTopicPartition;
+ }
+
+ private boolean isNewTopicPartition(final ByteRecord record) {
+ return lastTopicPartition == null
+ || !lastTopicPartition.getTopic().equals(record.getTopic())
+ || lastTopicPartition.getPartition() !=
record.getPartition();
+ }
+
+ private static long consumedBytes(final ByteRecord record) {
+ long recordSize = record.getValue().length;
+ final Optional<byte[]> key = record.getKey();
+ if (key.isPresent()) {
+ recordSize += key.get().length;
+ }
+ return recordSize;
+ }
+ }
+
private static class RebalanceSessionHolder implements SessionContext {
private final ProcessSession session;
private final OffsetTracker offsetTracker;
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/KafkaMetricName.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/KafkaMetricName.java
new file mode 100644
index 00000000000..8b099aac322
--- /dev/null
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/KafkaMetricName.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.kafka.processors.consumer;
+
+/**
+ * Metric names for ConsumeKafka counters and gauges.
+ */
+public enum KafkaMetricName {
+ BYTES_CONSUMED("kafka.bytes.consumed"),
+ RECORDS_CONSUMED("kafka.records.consumed"),
+ RECORDS_CONSUMER_LAG("kafka.records.consumer.lag"),
+ RECORDS_PARSED_ERRORS("kafka.records.parsed.errors");
+
+ private final String metricName;
+
+ KafkaMetricName(final String metricName) {
+ this.metricName = metricName;
+ }
+
+ public String getMetricName() {
+ return metricName;
+ }
+}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/OffsetTracker.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/OffsetTracker.java
index b79725940c1..a727140308b 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/OffsetTracker.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/OffsetTracker.java
@@ -24,11 +24,14 @@ import org.apache.nifi.kafka.service.api.record.ByteRecord;
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) {
@@ -36,13 +39,22 @@ public class OffsetTracker {
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);
+
+ long recordSize = consumerRecord.getValue().length;
+ final Optional<byte[]> key = consumerRecord.getKey();
+ if (key.isPresent()) {
+ recordSize += key.get().length;
+ }
+
+ totalRecordSize.addAndGet(recordSize);
+ }
+
+ public void recordConsumed(final TopicPartitionSummary topicPartition,
final long records, final long bytes) {
+ partitionRecords.merge(topicPartition, records, Long::sum);
+ partitionBytes.merge(topicPartition, bytes, Long::sum);
}
public long getTotalRecordSize() {
@@ -53,6 +65,14 @@ public class OffsetTracker {
return recordCounts;
}
+ public Map<TopicPartitionSummary, Long> getPartitionRecords() {
+ return partitionRecords;
+ }
+
+ public Map<TopicPartitionSummary, Long> getPartitionBytes() {
+ return partitionBytes;
+ }
+
public PollingSummary getPollingSummary(final PollingContext
pollingContext) {
final PollingSummary pollingSummary;
if (pollingContext.getTopicPattern().isPresent()) {
@@ -68,6 +88,8 @@ public class OffsetTracker {
public void clear() {
offsets.clear();
recordCounts.clear();
+ partitionRecords.clear();
+ partitionBytes.clear();
totalRecordSize.set(0);
}
}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/AbstractRecordStreamKafkaMessageConverter.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/AbstractRecordStreamKafkaMessageConverter.java
index bab7e0c93be..a707c22e34e 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/AbstractRecordStreamKafkaMessageConverter.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/AbstractRecordStreamKafkaMessageConverter.java
@@ -20,11 +20,14 @@ import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.kafka.processors.ConsumeKafka;
import org.apache.nifi.kafka.processors.common.HeaderValueConverter;
import org.apache.nifi.kafka.processors.common.KafkaUtils;
+import org.apache.nifi.kafka.processors.consumer.KafkaMetricName;
import org.apache.nifi.kafka.processors.consumer.OffsetTracker;
import org.apache.nifi.kafka.service.api.record.ByteRecord;
+import org.apache.nifi.kafka.shared.attribute.KafkaFlowFileAttribute;
import org.apache.nifi.kafka.shared.property.KeyEncoding;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.metrics.CommitTiming;
import org.apache.nifi.schema.access.SchemaNotFoundException;
import org.apache.nifi.serialization.MalformedRecordException;
import org.apache.nifi.serialization.RecordReader;
@@ -122,6 +125,11 @@ public abstract class
AbstractRecordStreamKafkaMessageConverter implements Kafka
ff = session.write(ff, out -> out.write(value));
session.transfer(ff, ConsumeKafka.PARSE_FAILURE);
session.adjustCounter("Records Received from " +
consumerRecord.getTopic(), 1, false);
+ final Map<String, String> counterAttributes = Map.of(
+ KafkaFlowFileAttribute.KAFKA_TOPIC, consumerRecord.getTopic(),
+ KafkaFlowFileAttribute.KAFKA_PARTITION,
Integer.toString(consumerRecord.getPartition())
+ );
+
session.adjustCounter(KafkaMetricName.RECORDS_PARSED_ERRORS.getMetricName(), 1,
counterAttributes, CommitTiming.NOW);
}
/**
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaTest.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaTest.java
index 92e82516415..4cd6fc81d71 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaTest.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaTest.java
@@ -19,9 +19,14 @@ package org.apache.nifi.kafka.processors;
import org.apache.nifi.components.Backlog;
import org.apache.nifi.components.ConfigVerificationResult;
import org.apache.nifi.components.ConfigVerificationResult.Outcome;
+import org.apache.nifi.kafka.processors.consumer.KafkaMetricName;
+import org.apache.nifi.kafka.processors.consumer.ProcessingStrategy;
import org.apache.nifi.kafka.service.api.KafkaConnectionService;
import org.apache.nifi.kafka.service.api.common.PartitionState;
+import org.apache.nifi.kafka.service.api.common.TopicPartitionSummary;
import org.apache.nifi.kafka.service.api.consumer.KafkaConsumerService;
+import org.apache.nifi.kafka.service.api.record.ByteRecord;
+import org.apache.nifi.kafka.shared.attribute.KafkaFlowFileAttribute;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
@@ -32,9 +37,13 @@ import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import java.lang.reflect.Field;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
import java.time.Instant;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.OptionalLong;
import java.util.Queue;
@@ -45,7 +54,9 @@ import static
org.apache.nifi.kafka.processors.ConsumeKafka.GROUP_ID;
import static org.apache.nifi.kafka.processors.ConsumeKafka.TOPICS;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.never;
@@ -59,6 +70,30 @@ class ConsumeKafkaTest {
private static final int FIRST_PARTITION = 0;
+ private static final int SECOND_PARTITION = 1;
+
+ private static final byte[] FIRST_RECORD_KEY =
"k".getBytes(StandardCharsets.UTF_8);
+
+ private static final byte[] FIRST_RECORD_VALUE =
"alpha".getBytes(StandardCharsets.UTF_8);
+
+ private static final byte[] SECOND_RECORD_VALUE =
"beta".getBytes(StandardCharsets.UTF_8);
+
+ private static final byte[] SECOND_RECORD_KEY =
"k2".getBytes(StandardCharsets.UTF_8);
+
+ private static final long FIRST_RECORD_BYTES = FIRST_RECORD_KEY.length +
FIRST_RECORD_VALUE.length;
+
+ private static final long SECOND_RECORD_BYTES = SECOND_RECORD_KEY.length +
SECOND_RECORD_VALUE.length;
+
+ private static final String RECORD_DEMARCATOR = "\n";
+
+ private static final long FIRST_PARTITION_LAG = 4L;
+
+ private static final long SECOND_PARTITION_LAG = 9L;
+
+ private static final String BROKER_URI = "kafka://localhost";
+
+ private static final String MAX_UNCOMMITTED_TIME = "250 ms";
+
private static final String DYNAMIC_PROPERTY_KEY_PUBLISH =
"delivery.timeout.ms";
private static final String DYNAMIC_PROPERTY_VALUE_PUBLISH = "60000";
private static final String DYNAMIC_PROPERTY_KEY_CONSUME =
"fetch.max.wait.ms";
@@ -231,6 +266,111 @@ class ConsumeKafkaTest {
assertEquals(OptionalLong.of(5L), afterStop.get().getRecordCount());
}
+ @Test
+ public void testOnTriggerRecordsBytesAndLagMetrics() throws
InitializationException {
+ setConnectionService();
+ configureProcessor();
+ when(kafkaConnectionService.getBrokerUri()).thenReturn(BROKER_URI);
+
when(kafkaConnectionService.getConsumerService(any())).thenReturn(kafkaConsumerService);
+
+ final ByteRecord firstRecord = new ByteRecord(TEST_TOPIC_NAME,
FIRST_PARTITION, 0, 0L, List.of(), FIRST_RECORD_KEY, FIRST_RECORD_VALUE, 1);
+ final ByteRecord secondRecord = new ByteRecord(TEST_TOPIC_NAME,
SECOND_PARTITION, 0, 0L, List.of(), null, SECOND_RECORD_VALUE, 1);
+
when(kafkaConsumerService.poll(any(Duration.class))).thenReturn(List.of(firstRecord,
secondRecord)).thenReturn(List.of());
+
+ final TopicPartitionSummary firstPartition = new
TopicPartitionSummary(TEST_TOPIC_NAME, FIRST_PARTITION);
+ final TopicPartitionSummary secondPartition = new
TopicPartitionSummary(TEST_TOPIC_NAME, SECOND_PARTITION);
+
when(kafkaConsumerService.currentLag(firstPartition)).thenReturn(OptionalLong.of(FIRST_PARTITION_LAG));
+
when(kafkaConsumerService.currentLag(secondPartition)).thenReturn(OptionalLong.of(SECOND_PARTITION_LAG));
+
+ runner.run();
+
+ final Map<String, String> firstAttributes =
topicPartitionAttributes(TEST_TOPIC_NAME, FIRST_PARTITION);
+ final Map<String, String> secondAttributes =
topicPartitionAttributes(TEST_TOPIC_NAME, SECOND_PARTITION);
+
+ assertEquals(1L,
runner.getCounterValue(KafkaMetricName.RECORDS_CONSUMED.getMetricName(),
firstAttributes));
+ assertEquals(1L,
runner.getCounterValue(KafkaMetricName.RECORDS_CONSUMED.getMetricName(),
secondAttributes));
+ assertEquals(FIRST_RECORD_BYTES,
runner.getCounterValue(KafkaMetricName.BYTES_CONSUMED.getMetricName(),
firstAttributes));
+ assertEquals(SECOND_RECORD_VALUE.length,
runner.getCounterValue(KafkaMetricName.BYTES_CONSUMED.getMetricName(),
secondAttributes));
+ assertEquals(2L, runner.getCounterValue("Records Received from " +
TEST_TOPIC_NAME));
+
+
assertFalse(runner.getGaugeValues(KafkaMetricName.RECORDS_CONSUMER_LAG.getMetricName(),
firstAttributes).isEmpty());
+
assertFalse(runner.getGaugeValues(KafkaMetricName.RECORDS_CONSUMER_LAG.getMetricName(),
secondAttributes).isEmpty());
+
assertTrue(runner.getGaugeValues(KafkaMetricName.RECORDS_CONSUMER_LAG.getMetricName(),
firstAttributes).stream().allMatch(value -> value == FIRST_PARTITION_LAG));
+
assertTrue(runner.getGaugeValues(KafkaMetricName.RECORDS_CONSUMER_LAG.getMetricName(),
secondAttributes).stream().allMatch(value -> value == SECOND_PARTITION_LAG));
+ runner.assertTransferCount(ConsumeKafka.SUCCESS, 2);
+ }
+
+ @Test
+ public void testOnTriggerEmptyPollDoesNotRecordLag() throws
InitializationException {
+ setConnectionService();
+ configureProcessor();
+ when(kafkaConnectionService.getBrokerUri()).thenReturn(BROKER_URI);
+
when(kafkaConnectionService.getConsumerService(any())).thenReturn(kafkaConsumerService);
+
when(kafkaConsumerService.poll(any(Duration.class))).thenReturn(List.of());
+
+ runner.run();
+
+ verify(kafkaConsumerService, never()).currentLag(any());
+ final Map<String, String> firstAttributes =
topicPartitionAttributes(TEST_TOPIC_NAME, FIRST_PARTITION);
+
assertTrue(runner.getGaugeValues(KafkaMetricName.RECORDS_CONSUMER_LAG.getMetricName(),
firstAttributes).isEmpty());
+ runner.assertTransferCount(ConsumeKafka.SUCCESS, 0);
+
assertNull(runner.getCounterValue(KafkaMetricName.RECORDS_CONSUMED.getMetricName(),
firstAttributes));
+ }
+
+ @Test
+ public void testOnTriggerEmitsConsumptionWhenPollFailsAfterRecords()
throws InitializationException {
+ setConnectionService();
+ configureProcessor();
+ when(kafkaConnectionService.getBrokerUri()).thenReturn(BROKER_URI);
+
when(kafkaConnectionService.getConsumerService(any())).thenReturn(kafkaConsumerService);
+
+ final ByteRecord firstRecord = new ByteRecord(TEST_TOPIC_NAME,
FIRST_PARTITION, 0, 0L, List.of(), FIRST_RECORD_KEY, FIRST_RECORD_VALUE, 1);
+
when(kafkaConsumerService.poll(any(Duration.class))).thenReturn(List.of(firstRecord)).thenThrow(new
RuntimeException("poll failed"));
+
+ runner.run();
+
+ final Map<String, String> firstAttributes =
topicPartitionAttributes(TEST_TOPIC_NAME, FIRST_PARTITION);
+ assertEquals(1L,
runner.getCounterValue(KafkaMetricName.RECORDS_CONSUMED.getMetricName(),
firstAttributes));
+ assertEquals(FIRST_RECORD_BYTES,
runner.getCounterValue(KafkaMetricName.BYTES_CONSUMED.getMetricName(),
firstAttributes));
+ runner.assertTransferCount(ConsumeKafka.SUCCESS, 0);
+ }
+
+ @Test
+ public void testOnTriggerDemarcatorRecordsOriginalKafkaBytes() throws
InitializationException {
+ setConnectionService();
+ configureProcessor();
+ runner.setProperty(ConsumeKafka.PROCESSING_STRATEGY,
ProcessingStrategy.DEMARCATOR);
+ runner.setProperty(ConsumeKafka.MESSAGE_DEMARCATOR, RECORD_DEMARCATOR);
+ when(kafkaConnectionService.getBrokerUri()).thenReturn(BROKER_URI);
+
when(kafkaConnectionService.getConsumerService(any())).thenReturn(kafkaConsumerService);
+
+ final ByteRecord firstRecord = new ByteRecord(TEST_TOPIC_NAME,
FIRST_PARTITION, 0, 0L, new ArrayList<>(), FIRST_RECORD_KEY,
FIRST_RECORD_VALUE, 1);
+ final ByteRecord secondRecord = new ByteRecord(TEST_TOPIC_NAME,
FIRST_PARTITION, 1, 0L, new ArrayList<>(), SECOND_RECORD_KEY,
SECOND_RECORD_VALUE, 1);
+
when(kafkaConsumerService.poll(any(Duration.class))).thenReturn(List.of(firstRecord,
secondRecord)).thenReturn(List.of());
+
+ final TopicPartitionSummary firstPartition = new
TopicPartitionSummary(TEST_TOPIC_NAME, FIRST_PARTITION);
+
when(kafkaConsumerService.currentLag(firstPartition)).thenReturn(OptionalLong.of(FIRST_PARTITION_LAG));
+
+ runner.run();
+
+ final Map<String, String> firstAttributes =
topicPartitionAttributes(TEST_TOPIC_NAME, FIRST_PARTITION);
+ final long originalKafkaBytes = FIRST_RECORD_BYTES +
SECOND_RECORD_BYTES;
+ final long bundledFlowFileBytes = FIRST_RECORD_VALUE.length
+ + RECORD_DEMARCATOR.getBytes(StandardCharsets.UTF_8).length
+ + SECOND_RECORD_VALUE.length;
+
+ assertEquals(2L,
runner.getCounterValue(KafkaMetricName.RECORDS_CONSUMED.getMetricName(),
firstAttributes));
+ assertEquals(originalKafkaBytes,
runner.getCounterValue(KafkaMetricName.BYTES_CONSUMED.getMetricName(),
firstAttributes));
+ assertNotEquals(bundledFlowFileBytes, originalKafkaBytes);
+ runner.assertTransferCount(ConsumeKafka.SUCCESS, 1);
+ }
+
+ private void configureProcessor() {
+ runner.setProperty(TOPICS, TEST_TOPIC_NAME);
+ runner.setProperty(GROUP_ID, CONSUMER_GROUP_ID);
+ runner.setProperty(ConsumeKafka.MAX_UNCOMMITTED_TIME,
MAX_UNCOMMITTED_TIME);
+ }
+
private void setConnectionService() throws InitializationException {
when(kafkaConnectionService.getIdentifier()).thenReturn(SERVICE_ID);
@@ -239,4 +379,11 @@ class ConsumeKafkaTest {
runner.setProperty(CONNECTION_SERVICE, SERVICE_ID);
}
+
+ private static Map<String, String> topicPartitionAttributes(final String
topic, final int partition) {
+ return Map.of(
+ KafkaFlowFileAttribute.KAFKA_TOPIC, topic,
+ KafkaFlowFileAttribute.KAFKA_PARTITION,
Integer.toString(partition)
+ );
+ }
}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/OffsetTrackerTest.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/OffsetTrackerTest.java
new file mode 100644
index 00000000000..995baae8fa2
--- /dev/null
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/OffsetTrackerTest.java
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.kafka.processors.consumer;
+
+import org.apache.nifi.kafka.service.api.common.TopicPartitionSummary;
+import org.apache.nifi.kafka.service.api.record.ByteRecord;
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class OffsetTrackerTest {
+
+ private static final String FIRST_TOPIC = "orders";
+ private static final String SECOND_TOPIC = "shipments";
+ private static final int FIRST_PARTITION = 0;
+ private static final int SECOND_PARTITION = 1;
+ private static final byte[] RECORD_KEY =
"key".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] FIRST_VALUE =
"alpha".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] SECOND_VALUE =
"beta".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] FIRST_KEY =
"k1".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] SECOND_KEY =
"k2".getBytes(StandardCharsets.UTF_8);
+ private static final long SECOND_TOPIC_BUNDLED_COUNT = 3L;
+ private static final byte[] NEWLINE_DEMARCATOR = new byte[]{10};
+
+ @Test
+ void testUpdateAggregatesRecordsAndBytesByPartition() {
+ final OffsetTracker offsetTracker = new OffsetTracker();
+ recordConsumed(offsetTracker, newByteRecord(FIRST_TOPIC,
FIRST_PARTITION, RECORD_KEY, FIRST_VALUE, 1));
+ recordConsumed(offsetTracker, newByteRecord(FIRST_TOPIC,
FIRST_PARTITION, null, SECOND_VALUE, 1));
+ recordConsumed(offsetTracker, newByteRecord(FIRST_TOPIC,
SECOND_PARTITION, RECORD_KEY, FIRST_VALUE, 1));
+ recordConsumed(offsetTracker, newByteRecord(SECOND_TOPIC,
FIRST_PARTITION, null, SECOND_VALUE, SECOND_TOPIC_BUNDLED_COUNT));
+ offsetTracker.update(newByteRecord(FIRST_TOPIC, FIRST_PARTITION,
RECORD_KEY, FIRST_VALUE, 1));
+ offsetTracker.update(newByteRecord(FIRST_TOPIC, FIRST_PARTITION, null,
SECOND_VALUE, 1));
+ offsetTracker.update(newByteRecord(FIRST_TOPIC, SECOND_PARTITION,
RECORD_KEY, FIRST_VALUE, 1));
+ offsetTracker.update(newByteRecord(SECOND_TOPIC, FIRST_PARTITION,
null, SECOND_VALUE, SECOND_TOPIC_BUNDLED_COUNT));
+
+ final TopicPartitionSummary firstTopicFirstPartition = new
TopicPartitionSummary(FIRST_TOPIC, FIRST_PARTITION);
+ final TopicPartitionSummary firstTopicSecondPartition = new
TopicPartitionSummary(FIRST_TOPIC, SECOND_PARTITION);
+ final TopicPartitionSummary secondTopicFirstPartition = new
TopicPartitionSummary(SECOND_TOPIC, FIRST_PARTITION);
+
+ final long firstTopicFirstPartitionBytes = RECORD_KEY.length +
FIRST_VALUE.length + SECOND_VALUE.length;
+ final long firstTopicSecondPartitionBytes = RECORD_KEY.length +
FIRST_VALUE.length;
+ final long secondTopicFirstPartitionBytes = SECOND_VALUE.length;
+ final long expectedTotalRecordSize = firstTopicFirstPartitionBytes +
firstTopicSecondPartitionBytes + secondTopicFirstPartitionBytes;
+
+ assertEquals(2L,
offsetTracker.getPartitionRecords().get(firstTopicFirstPartition));
+ assertEquals(1L,
offsetTracker.getPartitionRecords().get(firstTopicSecondPartition));
+ assertEquals(SECOND_TOPIC_BUNDLED_COUNT,
offsetTracker.getPartitionRecords().get(secondTopicFirstPartition));
+
+ assertEquals(firstTopicFirstPartitionBytes,
offsetTracker.getPartitionBytes().get(firstTopicFirstPartition));
+ assertEquals(firstTopicSecondPartitionBytes,
offsetTracker.getPartitionBytes().get(firstTopicSecondPartition));
+ assertEquals(secondTopicFirstPartitionBytes,
offsetTracker.getPartitionBytes().get(secondTopicFirstPartition));
+
+ assertEquals(3L, offsetTracker.getRecordCounts().get(FIRST_TOPIC));
+ assertEquals(SECOND_TOPIC_BUNDLED_COUNT,
offsetTracker.getRecordCounts().get(SECOND_TOPIC));
+ assertEquals(expectedTotalRecordSize,
offsetTracker.getTotalRecordSize());
+ }
+
+ @Test
+ void testRecordConsumedKeepsOriginalBytesWhenUpdateSeesBundledRecord() {
+ final OffsetTracker offsetTracker = new OffsetTracker();
+ final ByteRecord firstRecord = newByteRecord(FIRST_TOPIC,
FIRST_PARTITION, FIRST_KEY, FIRST_VALUE, 1);
+ final ByteRecord secondRecord = newByteRecord(FIRST_TOPIC,
FIRST_PARTITION, SECOND_KEY, SECOND_VALUE, 1);
+ final TopicPartitionSummary topicPartition = new
TopicPartitionSummary(FIRST_TOPIC, FIRST_PARTITION);
+ final long originalBytes = FIRST_KEY.length + FIRST_VALUE.length +
SECOND_KEY.length + SECOND_VALUE.length;
+
+ recordConsumed(offsetTracker, firstRecord);
+ recordConsumed(offsetTracker, secondRecord);
+
+ final byte[] bundledValue = new byte[FIRST_VALUE.length +
NEWLINE_DEMARCATOR.length + SECOND_VALUE.length];
+ System.arraycopy(FIRST_VALUE, 0, bundledValue, 0, FIRST_VALUE.length);
+ System.arraycopy(NEWLINE_DEMARCATOR, 0, bundledValue,
FIRST_VALUE.length, NEWLINE_DEMARCATOR.length);
+ System.arraycopy(SECOND_VALUE, 0, bundledValue, FIRST_VALUE.length +
NEWLINE_DEMARCATOR.length, SECOND_VALUE.length);
+ offsetTracker.update(newByteRecord(FIRST_TOPIC, FIRST_PARTITION, null,
bundledValue, 2));
+
+ assertEquals(2L,
offsetTracker.getPartitionRecords().get(topicPartition));
+ assertEquals(originalBytes,
offsetTracker.getPartitionBytes().get(topicPartition));
+ assertEquals(bundledValue.length, offsetTracker.getTotalRecordSize());
+ }
+
+ @Test
+ void testClearRemovesPartitionAggregates() {
+ final OffsetTracker offsetTracker = new OffsetTracker();
+ recordConsumed(offsetTracker, newByteRecord(FIRST_TOPIC,
FIRST_PARTITION, RECORD_KEY, FIRST_VALUE, 2));
+ offsetTracker.update(newByteRecord(FIRST_TOPIC, FIRST_PARTITION,
RECORD_KEY, FIRST_VALUE, 2));
+
+ offsetTracker.clear();
+
+ assertTrue(offsetTracker.getPartitionRecords().isEmpty());
+ assertTrue(offsetTracker.getPartitionBytes().isEmpty());
+ assertTrue(offsetTracker.getRecordCounts().isEmpty());
+ assertEquals(0L, offsetTracker.getTotalRecordSize());
+ }
+
+ private static void recordConsumed(final OffsetTracker offsetTracker,
final ByteRecord consumerRecord) {
+ offsetTracker.recordConsumed(
+ new TopicPartitionSummary(consumerRecord.getTopic(),
consumerRecord.getPartition()),
+ consumerRecord.getBundledCount(),
+ consumedBytes(consumerRecord));
+ }
+
+ private static long consumedBytes(final ByteRecord consumerRecord) {
+ long recordSize = consumerRecord.getValue().length;
+ if (consumerRecord.getKey().isPresent()) {
+ recordSize += consumerRecord.getKey().get().length;
+ }
+ return recordSize;
+ }
+
+ private static ByteRecord newByteRecord(final String topic, final int
partition, final byte[] key, final byte[] value, final long bundledCount) {
+ return new ByteRecord(topic, partition, 0, 0L, List.of(), key, value,
bundledCount);
+ }
+}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/RecordStreamKafkaMessageConverterTest.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/RecordStreamKafkaMessageConverterTest.java
index 8c544769de9..506f55517da 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/RecordStreamKafkaMessageConverterTest.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/RecordStreamKafkaMessageConverterTest.java
@@ -17,12 +17,16 @@
package org.apache.nifi.kafka.processors.consumer.convert;
import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.kafka.processors.consumer.KafkaMetricName;
import org.apache.nifi.kafka.processors.consumer.OffsetTracker;
import org.apache.nifi.kafka.service.api.record.ByteRecord;
import org.apache.nifi.kafka.shared.attribute.KafkaFlowFileAttribute;
import org.apache.nifi.kafka.shared.property.KeyEncoding;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.metrics.CommitTiming;
+import org.apache.nifi.serialization.MalformedRecordException;
+import org.apache.nifi.serialization.RecordReaderFactory;
import org.apache.nifi.serialization.RecordSetWriterFactory;
import org.apache.nifi.serialization.record.MockRecordParser;
import org.apache.nifi.serialization.record.RecordFieldType;
@@ -42,6 +46,8 @@ import java.util.regex.Pattern;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
@@ -50,6 +56,11 @@ import static org.mockito.Mockito.withSettings;
class RecordStreamKafkaMessageConverterTest {
+ private static final String FIRST_TOPIC = "topic1";
+ private static final String SECOND_TOPIC = "topic2";
+ private static final int FIRST_PARTITION = 0;
+ private static final String BROKER_URI = "brokerUri";
+
private final RecordSetWriterFactory writerFactory =
mock(RecordSetWriterFactory.class,
withSettings().defaultAnswer(Answers.RETURNS_DEEP_STUBS));
private final ComponentLog logger = mock(ComponentLog.class);
private final ProcessSession session = mock(ProcessSession.class,
withSettings().defaultAnswer(Answers.RETURNS_DEEP_STUBS));
@@ -78,17 +89,17 @@ class RecordStreamKafkaMessageConverterTest {
true,
offsetTracker,
logger,
- "brokerUri",
- new CreateNewFlowFileGrouping(writerFactory, logger,
"brokerUri", true)
+ BROKER_URI,
+ new CreateNewFlowFileGrouping(writerFactory, logger,
BROKER_URI, true)
);
// Create ByteRecords
- final ByteRecord group1Record1 = new ByteRecord("topic1", 0, 0, 1000L,
List.of(), null, "value1".getBytes(), 0L);
- final ByteRecord group1Record2 = new ByteRecord("topic1", 0, 3, 500L,
List.of(), null, "value4".getBytes(), 0L);
+ final ByteRecord group1Record1 = new ByteRecord(FIRST_TOPIC, 0, 0,
1000L, List.of(), null, "value1".getBytes(), 0L);
+ final ByteRecord group1Record2 = new ByteRecord(FIRST_TOPIC, 0, 3,
500L, List.of(), null, "value4".getBytes(), 0L);
- final ByteRecord group2 = new ByteRecord("topic1", 1, 1, 2000L,
List.of(), null, "value2".getBytes(), 0L);
+ final ByteRecord group2 = new ByteRecord(FIRST_TOPIC, 1, 1, 2000L,
List.of(), null, "value2".getBytes(), 0L);
- final ByteRecord group3 = new ByteRecord("topic2", 0, 2, 3000L,
List.of(), null, "value3".getBytes(), 0L);
+ final ByteRecord group3 = new ByteRecord(SECOND_TOPIC, 0, 2, 3000L,
List.of(), null, "value3".getBytes(), 0L);
final Iterator<ByteRecord> consumerRecords = List.of(group1Record1,
group2, group3, group1Record2).iterator();
// Mock the session.create() and session.write() methods
@@ -109,18 +120,18 @@ class RecordStreamKafkaMessageConverterTest {
final List<Map<String, String>> capturedAttributes =
attributesCaptor.getAllValues();
// check group1 records
- assertEquals("topic1",
capturedAttributes.get(0).get(KafkaFlowFileAttribute.KAFKA_TOPIC));
+ assertEquals(FIRST_TOPIC,
capturedAttributes.get(0).get(KafkaFlowFileAttribute.KAFKA_TOPIC));
assertEquals("0",
capturedAttributes.get(0).get(KafkaFlowFileAttribute.KAFKA_PARTITION));
- assertEquals("topic1",
capturedAttributes.get(3).get(KafkaFlowFileAttribute.KAFKA_TOPIC));
+ assertEquals(FIRST_TOPIC,
capturedAttributes.get(3).get(KafkaFlowFileAttribute.KAFKA_TOPIC));
assertEquals("0",
capturedAttributes.get(3).get(KafkaFlowFileAttribute.KAFKA_PARTITION));
//check group2 records
- assertEquals("topic1",
capturedAttributes.get(1).get(KafkaFlowFileAttribute.KAFKA_TOPIC));
+ assertEquals(FIRST_TOPIC,
capturedAttributes.get(1).get(KafkaFlowFileAttribute.KAFKA_TOPIC));
assertEquals("1",
capturedAttributes.get(1).get(KafkaFlowFileAttribute.KAFKA_PARTITION));
//check group3 records
- assertEquals("topic2",
capturedAttributes.get(2).get(KafkaFlowFileAttribute.KAFKA_TOPIC));
+ assertEquals(SECOND_TOPIC,
capturedAttributes.get(2).get(KafkaFlowFileAttribute.KAFKA_TOPIC));
assertEquals("0",
capturedAttributes.get(2).get(KafkaFlowFileAttribute.KAFKA_PARTITION));
final List<String> timestamps = capturedAttributes.stream()
@@ -131,4 +142,38 @@ class RecordStreamKafkaMessageConverterTest {
assertTrue(timestamps.contains("2000"), "Expected timestamp from
group2");
assertTrue(timestamps.contains("3000"), "Expected timestamp from
group3");
}
+
+ @Test
+ void testParseFailureIncrementsParseErrorCounterImmediately() throws
Exception {
+ final RecordReaderFactory readerFactory =
mock(RecordReaderFactory.class);
+ when(readerFactory.createRecordReader(any(), any(), anyLong(), any()))
+ .thenThrow(new MalformedRecordException("invalid record"));
+
+ final FlowFile flowFile = new MockFlowFile(1);
+ when(session.create()).thenReturn(flowFile);
+ when(session.putAllAttributes(any(FlowFile.class),
any())).thenReturn(flowFile);
+
when(session.write(any(FlowFile.class))).thenReturn(mock(OutputStream.class));
+
+ final RecordStreamKafkaMessageConverter converter = new
RecordStreamKafkaMessageConverter(
+ readerFactory,
+ writerFactory,
+ value -> new String(value, StandardCharsets.UTF_8),
+ Pattern.compile(".*"),
+ KeyEncoding.UTF8,
+ true,
+ offsetTracker,
+ logger,
+ BROKER_URI,
+ new CreateNewFlowFileGrouping(writerFactory, logger,
BROKER_URI, true)
+ );
+
+ final ByteRecord consumerRecord = new ByteRecord(FIRST_TOPIC,
FIRST_PARTITION, 0, 1000L, List.of(), null,
"invalid".getBytes(StandardCharsets.UTF_8), 1L);
+ converter.toFlowFiles(session, List.of(consumerRecord).iterator());
+
+
verify(session).adjustCounter(eq(KafkaMetricName.RECORDS_PARSED_ERRORS.getMetricName()),
eq(1L),
+ eq(Map.of(
+ KafkaFlowFileAttribute.KAFKA_TOPIC, FIRST_TOPIC,
+ KafkaFlowFileAttribute.KAFKA_PARTITION,
Integer.toString(FIRST_PARTITION))),
+ eq(CommitTiming.NOW));
+ }
}
