This is an automated email from the ASF dual-hosted git repository.
markap14 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 bb9e46c779d NIFI-16350 Skip Kafka tombstones in RECORD mode instead of
routing them to parse failure (#11857)
bb9e46c779d is described below
commit bb9e46c779d29094c2027687bcf418e28372124a
Author: Paweł Rzepiński <[email protected]>
AuthorDate: Mon Nov 9 14:58:52 2026 +0100
NIFI-16350 Skip Kafka tombstones in RECORD mode instead of routing them to
parse failure (#11857)
* NIFI-16350 Skip Kafka tombstones in ConsumeKafka RECORD mode
Null values were indistinguishable from empty payloads, so RECORD mode and
Verify treated tombstones as parse failures. Keep the distinction and skip only
real tombstones while still committing their offsets.
Co-authored-by: Cursor <[email protected]>
* NIFI-16350 Address review feedback on ConsumeKafka tombstone handling
Remove the ByteRecord constructor that defaulted tombstone to false.
Callers now pass the tombstone flag explicitly. Consolidate the
tombstone integration and converter tests, and rename test helpers
for clarity.
Co-authored-by: Cursor <[email protected]>
---------
Co-authored-by: Cursor <[email protected]>
---
.../kafka/processors/AbstractConsumeKafkaIT.java | 14 +++
.../processors/ConsumeKafkaRecordTombstoneIT.java | 135 +++++++++++++++++++++
.../apache/nifi/kafka/processors/ConsumeKafka.java | 26 +++-
.../consumer/bundle/ByteRecordBundler.java | 2 +-
.../AbstractRecordStreamKafkaMessageConverter.java | 7 ++
.../additionalDetails.md | 10 ++
.../nifi/kafka/processors/ConsumeKafkaTest.java | 92 +++++++++++++-
.../kafka/processors/common/KafkaUtilsTest.java | 3 +-
.../processors/consumer/OffsetTrackerTest.java | 2 +-
.../convert/CreateNewFlowFileGroupingTest.java | 16 +--
.../consumer/convert/MergeSchemaGroupingTest.java | 16 +--
.../RecordStreamKafkaMessageConverterTest.java | 27 +++--
...apperRecordStreamKafkaMessageConverterTest.java | 8 +-
.../nifi/kafka/service/api/record/ByteRecord.java | 11 +-
.../service/consumer/Kafka3ConsumerService.java | 12 +-
.../kafka/service/consumer/RecordIterable.java | 12 +-
.../consumer/Kafka3ConsumerServiceTest.java | 27 ++++-
17 files changed, 364 insertions(+), 56 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/AbstractConsumeKafkaIT.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/AbstractConsumeKafkaIT.java
index c4e481b7822..6e4edd169c4 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/AbstractConsumeKafkaIT.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/AbstractConsumeKafkaIT.java
@@ -21,6 +21,7 @@ import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.nifi.util.TestRunner;
@@ -60,6 +61,19 @@ public abstract class AbstractConsumeKafkaIT extends
AbstractKafkaBaseIT {
}
}
+ protected void produceBytes(final String topic, final int partition, final
String key, final byte[] value)
+ throws ExecutionException, InterruptedException {
+ final Properties properties = getKafkaProducerProperties();
+ properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
ByteArraySerializer.class.getName());
+
+ try (final KafkaProducer<String, byte[]> producer = new
KafkaProducer<>(properties)) {
+ final ProducerRecord<String, byte[]> record = new
ProducerRecord<>(topic, partition, key, value);
+ final RecordMetadata metadata = producer.send(record).get();
+ assertEquals(topic, metadata.topic());
+ assertTrue(metadata.hasOffset());
+ }
+ }
+
protected void produce(final String topic, final
Collection<ProducerRecord<String, String>> records)
throws ExecutionException, InterruptedException {
final Collection<RecordMetadata> metadatas = new ArrayList<>();
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaRecordTombstoneIT.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaRecordTombstoneIT.java
new file mode 100644
index 00000000000..1a35de4394b
--- /dev/null
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaRecordTombstoneIT.java
@@ -0,0 +1,135 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.avro.Schema;
+import org.apache.avro.file.DataFileWriter;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.nifi.avro.AvroReader;
+import org.apache.nifi.kafka.processors.consumer.ProcessingStrategy;
+import org.apache.nifi.kafka.service.api.consumer.AutoOffsetReset;
+import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.schema.access.SchemaAccessUtils;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.time.Duration;
+import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+
+/**
+ * RECORD output for a batch that contains a non-null empty value, a null
tombstone, and a real Avro value.
+ * The tombstone is omitted from success and parse.failure. The empty value is
routed to parse.failure.
+ * The real value is written to success.
+ */
+class ConsumeKafkaRecordTombstoneIT extends AbstractConsumeKafkaIT {
+
+ private static final byte[] EMPTY_BYTES = new byte[0];
+
+ private static final String AVRO_SCHEMA = """
+ {
+ "type": "record",
+ "name": "message",
+ "fields": [
+ {"name": "name", "type": "string"}
+ ]
+ }
+ """;
+
+ private static final String MESSAGE_KEY = "key";
+
+ private static final int FIRST_PARTITION = 0;
+
+ private static final Duration POLL_TIMEOUT = Duration.ofSeconds(30);
+
+ private TestRunner runner;
+
+ private String topic;
+
+ @BeforeEach
+ void setRunner() throws InitializationException {
+ runner = TestRunners.newTestRunner(ConsumeKafka.class);
+ addKafkaConnectionService(runner);
+ runner.setProperty(ConsumeKafka.CONNECTION_SERVICE,
CONNECTION_SERVICE_ID);
+ addRecordWriterService(runner);
+ useAvroReader();
+
+ topic = UUID.randomUUID().toString();
+ runner.setProperty(ConsumeKafka.GROUP_ID, topic);
+ runner.setProperty(ConsumeKafka.TOPICS, topic);
+ runner.setProperty(ConsumeKafka.PROCESSING_STRATEGY,
ProcessingStrategy.RECORD);
+ runner.setProperty(ConsumeKafka.AUTO_OFFSET_RESET,
AutoOffsetReset.EARLIEST.getValue());
+ }
+
+ @Test
+ void testTombstoneSkippedAndEmptyValueRoutedToParseFailure() throws
ExecutionException, InterruptedException, IOException {
+ produceAndConsumeAvroRecords();
+
+ runner.assertTransferCount(ConsumeKafka.PARSE_FAILURE, 1);
+ runner.assertTransferCount(ConsumeKafka.SUCCESS, 1);
+ final MockFlowFile flowFile =
runner.getFlowFilesForRelationship(ConsumeKafka.SUCCESS).getFirst();
+ final ArrayNode records = assertInstanceOf(ArrayNode.class,
objectMapper.readTree(flowFile.getContent()));
+ assertEquals(1, records.size());
+ final ObjectNode record = assertInstanceOf(ObjectNode.class,
records.get(0));
+ assertEquals("A", record.get("name").asText());
+ }
+
+ private void produceAndConsumeAvroRecords() throws ExecutionException,
InterruptedException, IOException {
+ runner.run(1, false, true);
+ produceBytes(topic, FIRST_PARTITION, MESSAGE_KEY, EMPTY_BYTES);
+ produceBytes(topic, FIRST_PARTITION, MESSAGE_KEY, null);
+ produceBytes(topic, FIRST_PARTITION, MESSAGE_KEY, createAvroRecord());
+ runUntil(runner, testRunner ->
!testRunner.getFlowFilesForRelationship(ConsumeKafka.SUCCESS).isEmpty()
+ &&
!testRunner.getFlowFilesForRelationship(ConsumeKafka.PARSE_FAILURE).isEmpty(),
POLL_TIMEOUT);
+ runner.run(1, true, false);
+ }
+
+ private void useAvroReader() throws InitializationException {
+ final String readerId = "Avro Reader";
+ final AvroReader reader = new AvroReader();
+ runner.addControllerService(readerId, reader);
+ runner.setProperty(reader, SchemaAccessUtils.SCHEMA_ACCESS_STRATEGY,
"embedded-avro-schema");
+ runner.enableControllerService(reader);
+ runner.setProperty(ConsumeKafka.RECORD_READER, readerId);
+ }
+
+ private byte[] createAvroRecord() throws IOException {
+ final Schema schema = new Schema.Parser().parse(AVRO_SCHEMA);
+ final GenericRecord record = new GenericData.Record(schema);
+ record.put("name", "A");
+
+ final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ try (DataFileWriter<GenericRecord> writer = new DataFileWriter<>(new
GenericDatumWriter<>(schema))) {
+ writer.create(schema, outputStream);
+ writer.append(record);
+ }
+
+ return outputStream.toByteArray();
+ }
+}
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 f000ea8b8be..4e532eb4c21 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
@@ -515,6 +515,7 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
getLogger().debug("Rebalance detected with revoked
partitions, breaking to commit session");
break;
}
+
continue;
}
@@ -570,6 +571,7 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
} catch (final Exception e) {
getLogger().warn("Failed to commit offsets for revoked
partitions", e);
}
+
consumerServices.offer(consumerService);
return;
}
@@ -791,8 +793,15 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
final RecordReaderFactory readerFactory =
context.getProperty(RECORD_READER).asControllerService(RecordReaderFactory.class);
int recordIndex = 0;
+ int parsedRecords = 0;
+ int skippedTombstones = 0;
for (final ByteRecord byteRecord : records) {
recordIndex++;
+ if (byteRecord.isTombstone()) {
+ skippedTombstones++;
+ continue;
+ }
+
final Map<String, String> recordAttributes =
KafkaUtils.toAttributes(
byteRecord, keyEncoding, headerNamePattern,
headerValueConverter, commitOffsets);
@@ -808,24 +817,31 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
.explanation("Failed to parse Record number " +
recordIndex + ": " + e)
.build();
}
+
+ parsedRecords++;
}
- if (recordIndex == 0) {
+ if (parsedRecords == 0) {
+ final String explanation = skippedTombstones == 0
+ ? "No records were received to parse"
+ : "No records were parsed (skipped " + skippedTombstones +
" tombstones)";
return new ConfigVerificationResult.Builder()
.verificationStepName("Parse Records")
.outcome(Outcome.SKIPPED)
- .explanation("No records were received to parse")
+ .explanation(explanation)
.build();
}
+ final String explanation = skippedTombstones == 0
+ ? "Successfully parsed " + parsedRecords + " records"
+ : "Successfully parsed " + parsedRecords + " records (skipped
" + skippedTombstones + " tombstones)";
return new ConfigVerificationResult.Builder()
.verificationStepName("Parse Records")
.outcome(Outcome.SUCCESSFUL)
- .explanation("Successfully parsed " + recordIndex + " records")
+ .explanation(explanation)
.build();
}
-
@ConnectorMethod(
name = "sampleTopics",
description = "Returns a list of sample data from the topics that
would be consumed by this processor."
@@ -1032,6 +1048,7 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
topicPartitionSummaries.add(summary);
lastTopicPartition = summary;
}
+
return lastTopicPartition;
}
@@ -1047,6 +1064,7 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
if (key.isPresent()) {
recordSize += key.get().length;
}
+
return recordSize;
}
}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/bundle/ByteRecordBundler.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/bundle/ByteRecordBundler.java
index c47e308cfe6..c3944ced572 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/bundle/ByteRecordBundler.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/bundle/ByteRecordBundler.java
@@ -66,7 +66,7 @@ public class ByteRecordBundler {
key.headers.add(new
RecordHeader(KafkaFlowFileAttribute.KAFKA_MAX_OFFSET,
Long.toString(value.getLastOffset()).getBytes(StandardCharsets.UTF_8)));
key.headers.add(new RecordHeader(KafkaFlowFileAttribute.KAFKA_COUNT,
Long.toString(value.getCount()).getBytes(StandardCharsets.UTF_8)));
return new ByteRecord(topicPartition.getTopic(),
topicPartition.getPartition(),
- value.getFirstOffset(), key.getTimestamp(), key.getHeaders(),
key.getMessageKey(), value.getData(), value.getCount());
+ value.getFirstOffset(), key.getTimestamp(), key.getHeaders(),
key.getMessageKey(), value.getData(), value.getCount(), false);
}
private void update(final Map<BundleKey, BundleValue> bundles, final
ByteRecord byteRecord) {
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 a707c22e34e..5f9d65c7009 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
@@ -87,6 +87,13 @@ public abstract class
AbstractRecordStreamKafkaMessageConverter implements Kafka
public void toFlowFiles(final ProcessSession session, final
Iterator<ByteRecord> consumerRecords) {
while (consumerRecords.hasNext()) {
final ByteRecord consumerRecord = consumerRecords.next();
+ if (consumerRecord.isTombstone()) {
+ session.adjustCounter("Records Received from " +
consumerRecord.getTopic(), 1, false);
+ session.adjustCounter("Tombstones Skipped for " +
consumerRecord.getTopic(), 1, false);
+ offsetTracker.update(consumerRecord);
+ continue;
+ }
+
final byte[] value = consumerRecord.getValue();
final Map<String, String> attributes = KafkaUtils.toAttributes(
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/resources/docs/org.apache.nifi.kafka.processors.ConsumeKafka/additionalDetails.md
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/resources/docs/org.apache.nifi.kafka.processors.ConsumeKafka/additionalDetails.md
index fb5f455cd16..bd9b8176095 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/resources/docs/org.apache.nifi.kafka.processors.ConsumeKafka/additionalDetails.md
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/resources/docs/org.apache.nifi.kafka.processors.ConsumeKafka/additionalDetails.md
@@ -176,6 +176,16 @@ Here is an example of FlowFile content that is emitted by
JsonRecordSetWriter wh
]
```
+### Handling tombstones
+
+A Kafka tombstone is a record whose value is null (which is not the same as
empty byte array).
+
+Handling depends on Processing Strategy:
+
+- FLOW_FILE: The tombstone is represented as a FlowFile with `kafka.tombstone`
attribute set to `true`.
+- DEMARCATOR: The tombstone is included in the bundle records as empty byte
array.
+- RECORD: The tombstone is skipped for every Output Strategy. The processor
increments `Tombstones Skipped for <topic>` counter.
+
### Memory Usage with Continue with Merged Schema
When the Schema Conflict Resolution property is set to 'Continue with Merged
Schema', the records from an entire poll are grouped by topic and partition and
held in memory until they are written to FlowFiles.
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 4cd6fc81d71..07bfa326c57 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
@@ -28,6 +28,8 @@ 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.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.junit.jupiter.api.BeforeEach;
@@ -36,6 +38,7 @@ import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
+import java.io.IOException;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
@@ -59,6 +62,8 @@ 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.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -101,6 +106,8 @@ class ConsumeKafkaTest {
private static final String SERVICE_ID =
KafkaConnectionService.class.getSimpleName();
+ private static final String READER_ID = "record-reader";
+
private static final String CONSUMER_GROUP_ID =
ConsumeKafkaTest.class.getSimpleName();
@Mock
@@ -177,6 +184,55 @@ class ConsumeKafkaTest {
assertNotNull(firstResult.getExplanation());
}
+ @Test
+ public void testVerifySkipsTombstoneAndParsesFollowingRecord() throws
Exception {
+ final RecordReaderFactory readerFactory = configureRecordReader();
+ final RecordReader recordReader = mock(RecordReader.class);
+ when(readerFactory.createRecordReader(any(), any(), anyLong(),
any())).thenAnswer(invocation -> {
+ final long length = invocation.getArgument(2);
+ if (length == 0L) {
+ throw new IOException("empty value");
+ }
+
+ return recordReader;
+ });
+
+ final ByteRecord tombstone = new ByteRecord(TEST_TOPIC_NAME,
FIRST_PARTITION, 0L, 0L, List.of(), null, new byte[0], 1, true);
+ final ByteRecord dataRecord = new ByteRecord(TEST_TOPIC_NAME,
FIRST_PARTITION, 1L, 0L, List.of(), null, FIRST_RECORD_VALUE, 1, false);
+
when(kafkaConsumerService.poll(any(Duration.class))).thenReturn(List.of(tombstone,
dataRecord));
+
+ final ConfigVerificationResult parseResult =
findParseResult(verifyRecords());
+ assertEquals(Outcome.SUCCESSFUL, parseResult.getOutcome());
+ assertEquals("Successfully parsed 1 records (skipped 1 tombstones)",
parseResult.getExplanation());
+ }
+
+ @Test
+ public void testVerifySkipsWhenSampleContainsOnlyTombstones() throws
Exception {
+ final RecordReaderFactory readerFactory = configureRecordReader();
+ final ByteRecord tombstone = new ByteRecord(TEST_TOPIC_NAME,
FIRST_PARTITION, 0L, 0L, List.of(), null, new byte[0], 1, true);
+
when(kafkaConsumerService.poll(any(Duration.class))).thenReturn(List.of(tombstone));
+
+ final ConfigVerificationResult parseResult =
findParseResult(verifyRecords());
+ assertEquals(Outcome.SKIPPED, parseResult.getOutcome());
+ assertEquals("No records were parsed (skipped 1 tombstones)",
parseResult.getExplanation());
+ verify(readerFactory, never()).createRecordReader(any(), any(),
anyLong(), any());
+ }
+
+ @Test
+ public void testVerifyReportsDataRecordWhenTombstonePrecedesParseFailure()
throws Exception {
+ final RecordReaderFactory readerFactory = configureRecordReader();
+ when(readerFactory.createRecordReader(any(), any(), anyLong(),
any())).thenThrow(new IOException("empty value"));
+
+ final ByteRecord tombstone = new ByteRecord(TEST_TOPIC_NAME,
FIRST_PARTITION, 0L, 0L, List.of(), null, new byte[0], 1, true);
+ final ByteRecord dataRecord = new ByteRecord(TEST_TOPIC_NAME,
FIRST_PARTITION, 1L, 0L, List.of(), null, FIRST_RECORD_VALUE, 1, false);
+
when(kafkaConsumerService.poll(any(Duration.class))).thenReturn(List.of(tombstone,
dataRecord));
+
+ final ConfigVerificationResult parseResult =
findParseResult(verifyRecords());
+ assertEquals(Outcome.FAILED, parseResult.getOutcome());
+ assertEquals("Failed to parse Record number 2: java.io.IOException:
empty value", parseResult.getExplanation());
+ verify(readerFactory).createRecordReader(any(), any(), anyLong(),
any());
+ }
+
@Test
public void testDynamicProperties() throws InitializationException {
when(kafkaConnectionService.getIdentifier()).thenReturn(SERVICE_ID);
@@ -273,8 +329,8 @@ class ConsumeKafkaTest {
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);
+ final ByteRecord firstRecord = new ByteRecord(TEST_TOPIC_NAME,
FIRST_PARTITION, 0, 0L, List.of(), FIRST_RECORD_KEY, FIRST_RECORD_VALUE, 1,
false);
+ final ByteRecord secondRecord = new ByteRecord(TEST_TOPIC_NAME,
SECOND_PARTITION, 0, 0L, List.of(), null, SECOND_RECORD_VALUE, 1, false);
when(kafkaConsumerService.poll(any(Duration.class))).thenReturn(List.of(firstRecord,
secondRecord)).thenReturn(List.of());
final TopicPartitionSummary firstPartition = new
TopicPartitionSummary(TEST_TOPIC_NAME, FIRST_PARTITION);
@@ -324,7 +380,7 @@ class ConsumeKafkaTest {
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 firstRecord = new ByteRecord(TEST_TOPIC_NAME,
FIRST_PARTITION, 0, 0L, List.of(), FIRST_RECORD_KEY, FIRST_RECORD_VALUE, 1,
false);
when(kafkaConsumerService.poll(any(Duration.class))).thenReturn(List.of(firstRecord)).thenThrow(new
RuntimeException("poll failed"));
runner.run();
@@ -344,8 +400,8 @@ class ConsumeKafkaTest {
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);
+ final ByteRecord firstRecord = new ByteRecord(TEST_TOPIC_NAME,
FIRST_PARTITION, 0, 0L, new ArrayList<>(), FIRST_RECORD_KEY,
FIRST_RECORD_VALUE, 1, false);
+ final ByteRecord secondRecord = new ByteRecord(TEST_TOPIC_NAME,
FIRST_PARTITION, 1, 0L, new ArrayList<>(), SECOND_RECORD_KEY,
SECOND_RECORD_VALUE, 1, false);
when(kafkaConsumerService.poll(any(Duration.class))).thenReturn(List.of(firstRecord,
secondRecord)).thenReturn(List.of());
final TopicPartitionSummary firstPartition = new
TopicPartitionSummary(TEST_TOPIC_NAME, FIRST_PARTITION);
@@ -365,6 +421,32 @@ class ConsumeKafkaTest {
runner.assertTransferCount(ConsumeKafka.SUCCESS, 1);
}
+ private RecordReaderFactory configureRecordReader() throws
InitializationException {
+ final RecordReaderFactory readerFactory =
mock(RecordReaderFactory.class);
+ when(readerFactory.getIdentifier()).thenReturn(READER_ID);
+ runner.addControllerService(READER_ID, readerFactory);
+ runner.enableControllerService(readerFactory);
+ runner.setProperty(ConsumeKafka.PROCESSING_STRATEGY,
ProcessingStrategy.RECORD);
+ runner.setProperty(ConsumeKafka.RECORD_READER, READER_ID);
+ return readerFactory;
+ }
+
+ private List<ConfigVerificationResult> verifyRecords() throws
InitializationException {
+ when(kafkaConsumerService.getPartitionStates()).thenReturn(List.of(new
PartitionState(TEST_TOPIC_NAME, FIRST_PARTITION)));
+ setConnectionService();
+
when(kafkaConnectionService.getConsumerService(any())).thenReturn(kafkaConsumerService);
+ runner.setProperty(TOPICS, TEST_TOPIC_NAME);
+ runner.setProperty(GROUP_ID, CONSUMER_GROUP_ID);
+ return processor.verify(runner.getProcessContext(),
runner.getLogger(), Collections.emptyMap());
+ }
+
+ private static ConfigVerificationResult findParseResult(final
List<ConfigVerificationResult> results) {
+ return results.stream()
+ .filter(result -> "Parse
Records".equals(result.getVerificationStepName()))
+ .findFirst()
+ .orElseThrow();
+ }
+
private void configureProcessor() {
runner.setProperty(TOPICS, TEST_TOPIC_NAME);
runner.setProperty(GROUP_ID, CONSUMER_GROUP_ID);
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/common/KafkaUtilsTest.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/common/KafkaUtilsTest.java
index e319fd44a4f..aea888d4636 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/common/KafkaUtilsTest.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/common/KafkaUtilsTest.java
@@ -189,7 +189,8 @@ class KafkaUtilsTest {
headers,
null, // key
"test-value".getBytes(StandardCharsets.UTF_8),
- 1 // count
+ 1,
+ false
);
}
}
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
index 995baae8fa2..92cbecf8f22 100644
---
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
@@ -126,6 +126,6 @@ class OffsetTrackerTest {
}
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);
+ return new ByteRecord(topic, partition, 0, 0L, List.of(), key, value,
bundledCount, false);
}
}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/CreateNewFlowFileGroupingTest.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/CreateNewFlowFileGroupingTest.java
index f459ae18f3d..789e08445ea 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/CreateNewFlowFileGroupingTest.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/CreateNewFlowFileGroupingTest.java
@@ -82,8 +82,8 @@ class CreateNewFlowFileGroupingTest {
@Test
void testSameTopicPartitionAndSchemaShareOneFlowFile() throws Exception {
- final ByteRecord first = new ByteRecord(TOPIC, 0, 10, 1000L,
List.of(), null, new byte[0], 0L);
- final ByteRecord second = new ByteRecord(TOPIC, 0, 11, 500L,
List.of(), null, new byte[0], 0L);
+ final ByteRecord first = new ByteRecord(TOPIC, 0, 10, 1000L,
List.of(), null, new byte[0], 0L, false);
+ final ByteRecord second = new ByteRecord(TOPIC, 0, 11, 500L,
List.of(), null, new byte[0], 0L, false);
grouping.addRecord(session, first, RECORD_A, SCHEMA_A, Map.of(),
Map.of());
grouping.addRecord(session, second, RECORD_A, SCHEMA_A, Map.of(),
Map.of());
@@ -104,8 +104,8 @@ class CreateNewFlowFileGroupingTest {
@Test
void testDifferentWriteSchemasProduceSeparateFlowFiles() throws Exception {
- final ByteRecord first = new ByteRecord(TOPIC, 0, 1, 1000L, List.of(),
null, new byte[0], 0L);
- final ByteRecord second = new ByteRecord(TOPIC, 0, 2, 2000L,
List.of(), null, new byte[0], 0L);
+ final ByteRecord first = new ByteRecord(TOPIC, 0, 1, 1000L, List.of(),
null, new byte[0], 0L, false);
+ final ByteRecord second = new ByteRecord(TOPIC, 0, 2, 2000L,
List.of(), null, new byte[0], 0L, false);
grouping.addRecord(session, first, RECORD_A, SCHEMA_A, Map.of(),
Map.of());
grouping.addRecord(session, second, RECORD_B, SCHEMA_B, Map.of(),
Map.of());
@@ -119,8 +119,8 @@ class CreateNewFlowFileGroupingTest {
@Test
void testDifferentGroupingAttributesProduceSeparateFlowFiles() throws
Exception {
- final ByteRecord first = new ByteRecord(TOPIC, 0, 1, 1000L, List.of(),
null, new byte[0], 0L);
- final ByteRecord second = new ByteRecord(TOPIC, 0, 2, 2000L,
List.of(), null, new byte[0], 0L);
+ final ByteRecord first = new ByteRecord(TOPIC, 0, 1, 1000L, List.of(),
null, new byte[0], 0L, false);
+ final ByteRecord second = new ByteRecord(TOPIC, 0, 2, 2000L,
List.of(), null, new byte[0], 0L, false);
grouping.addRecord(session, first, RECORD_A, SCHEMA_A, Map.of(),
Map.of("hdr", "a"));
grouping.addRecord(session, second, RECORD_A, SCHEMA_A, Map.of(),
Map.of("hdr", "b"));
@@ -134,8 +134,8 @@ class CreateNewFlowFileGroupingTest {
@Test
void testDifferentPartitionsProduceSeparateFlowFiles() throws Exception {
- final ByteRecord first = new ByteRecord(TOPIC, 0, 1, 1000L, List.of(),
null, new byte[0], 0L);
- final ByteRecord second = new ByteRecord(TOPIC, 1, 2, 2000L,
List.of(), null, new byte[0], 0L);
+ final ByteRecord first = new ByteRecord(TOPIC, 0, 1, 1000L, List.of(),
null, new byte[0], 0L, false);
+ final ByteRecord second = new ByteRecord(TOPIC, 1, 2, 2000L,
List.of(), null, new byte[0], 0L, false);
grouping.addRecord(session, first, RECORD_A, SCHEMA_A, Map.of(),
Map.of());
grouping.addRecord(session, second, RECORD_A, SCHEMA_A, Map.of(),
Map.of());
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/MergeSchemaGroupingTest.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/MergeSchemaGroupingTest.java
index a98a782bc1e..543341ebd81 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/MergeSchemaGroupingTest.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/MergeSchemaGroupingTest.java
@@ -86,8 +86,8 @@ class MergeSchemaGroupingTest {
@Test
void testDifferentSchemasInSameTopicPartitionMergeIntoOneFlowFile() {
- final ByteRecord first = new ByteRecord(TOPIC, 0, 10, 1000L,
List.of(), null, new byte[0], 0L);
- final ByteRecord second = new ByteRecord(TOPIC, 0, 11, 500L,
List.of(), null, new byte[0], 0L);
+ final ByteRecord first = new ByteRecord(TOPIC, 0, 10, 1000L,
List.of(), null, new byte[0], 0L, false);
+ final ByteRecord second = new ByteRecord(TOPIC, 0, 11, 500L,
List.of(), null, new byte[0], 0L, false);
grouping.addRecord(session, first, RECORD_A, SCHEMA_A, Map.of(),
Map.of());
grouping.addRecord(session, second, RECORD_B, SCHEMA_B, Map.of(),
Map.of());
@@ -110,8 +110,8 @@ class MergeSchemaGroupingTest {
@Test
void testDifferentPartitionsProduceSeparateFlowFiles() {
- final ByteRecord first = new ByteRecord(TOPIC, 0, 1, 1000L, List.of(),
null, new byte[0], 0L);
- final ByteRecord second = new ByteRecord(TOPIC, 1, 2, 2000L,
List.of(), null, new byte[0], 0L);
+ final ByteRecord first = new ByteRecord(TOPIC, 0, 1, 1000L, List.of(),
null, new byte[0], 0L, false);
+ final ByteRecord second = new ByteRecord(TOPIC, 1, 2, 2000L,
List.of(), null, new byte[0], 0L, false);
grouping.addRecord(session, first, RECORD_A, SCHEMA_A, Map.of(),
Map.of());
grouping.addRecord(session, second, RECORD_B, SCHEMA_B, Map.of(),
Map.of());
@@ -125,8 +125,8 @@ class MergeSchemaGroupingTest {
@Test
void testDifferentGroupingAttributesProduceSeparateFlowFiles() {
- final ByteRecord first = new ByteRecord(TOPIC, 0, 1, 1000L, List.of(),
null, new byte[0], 0L);
- final ByteRecord second = new ByteRecord(TOPIC, 0, 2, 2000L,
List.of(), null, new byte[0], 0L);
+ final ByteRecord first = new ByteRecord(TOPIC, 0, 1, 1000L, List.of(),
null, new byte[0], 0L, false);
+ final ByteRecord second = new ByteRecord(TOPIC, 0, 2, 2000L,
List.of(), null, new byte[0], 0L, false);
grouping.addRecord(session, first, RECORD_A, SCHEMA_A, Map.of(),
Map.of("hdr", "a"));
grouping.addRecord(session, second, RECORD_B, SCHEMA_B, Map.of(),
Map.of("hdr", "b"));
@@ -160,8 +160,8 @@ class MergeSchemaGroupingTest {
final MergeSchemaGrouping validatingGrouping = new MergeSchemaGrouping(
validatingWriter, runner.getLogger(), BROKER_URI, true);
- final ByteRecord first = new ByteRecord(TOPIC, 0, 10, 1000L,
List.of(), null, new byte[0], 0L);
- final ByteRecord second = new ByteRecord(TOPIC, 0, 11, 2000L,
List.of(), null, new byte[0], 0L);
+ final ByteRecord first = new ByteRecord(TOPIC, 0, 10, 1000L,
List.of(), null, new byte[0], 0L, false);
+ final ByteRecord second = new ByteRecord(TOPIC, 0, 11, 2000L,
List.of(), null, new byte[0], 0L, false);
validatingGrouping.addRecord(validatingSession, first, recordA,
nonNullableSchemaA, Map.of(), Map.of());
validatingGrouping.addRecord(validatingSession, second, recordB,
nonNullableSchemaB, Map.of(), Map.of());
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 506f55517da..91cf8ad662c 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,6 +17,7 @@
package org.apache.nifi.kafka.processors.consumer.convert;
import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.kafka.processors.ConsumeKafka;
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;
@@ -24,6 +25,7 @@ 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.io.OutputStreamCallback;
import org.apache.nifi.processor.metrics.CommitTiming;
import org.apache.nifi.serialization.MalformedRecordException;
import org.apache.nifi.serialization.RecordReaderFactory;
@@ -50,6 +52,7 @@ 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.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.mockito.Mockito.withSettings;
@@ -94,12 +97,12 @@ class RecordStreamKafkaMessageConverterTest {
);
// Create ByteRecords
- 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 group1Record1 = new ByteRecord(FIRST_TOPIC, 0, 0,
1000L, List.of(), null, "value1".getBytes(), 0L, false);
+ final ByteRecord group1Record2 = new ByteRecord(FIRST_TOPIC, 0, 3,
500L, List.of(), null, "value4".getBytes(), 0L, false);
- final ByteRecord group2 = new ByteRecord(FIRST_TOPIC, 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, false);
- final ByteRecord group3 = new ByteRecord(SECOND_TOPIC, 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, false);
final Iterator<ByteRecord> consumerRecords = List.of(group1Record1,
group2, group3, group1Record2).iterator();
// Mock the session.create() and session.write() methods
@@ -144,15 +147,15 @@ class RecordStreamKafkaMessageConverterTest {
}
@Test
- void testParseFailureIncrementsParseErrorCounterImmediately() throws
Exception {
+ void testTombstoneSkippedAndEmptyValueIncrementsParseErrorCounter() throws
Exception {
final RecordReaderFactory readerFactory =
mock(RecordReaderFactory.class);
when(readerFactory.createRecordReader(any(), any(), anyLong(), any()))
- .thenThrow(new MalformedRecordException("invalid record"));
+ .thenThrow(new MalformedRecordException("empty value"));
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));
+ when(session.write(any(FlowFile.class),
any(OutputStreamCallback.class))).thenReturn(flowFile);
final RecordStreamKafkaMessageConverter converter = new
RecordStreamKafkaMessageConverter(
readerFactory,
@@ -167,13 +170,19 @@ class RecordStreamKafkaMessageConverterTest {
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());
+ final ByteRecord tombstone = new ByteRecord(FIRST_TOPIC,
FIRST_PARTITION, 4L, 1000L, List.of(), null, new byte[0], 1L, true);
+ final ByteRecord emptyValue = new ByteRecord(FIRST_TOPIC,
FIRST_PARTITION, 5L, 1000L, List.of(), null, new byte[0], 1L, false);
+ converter.toFlowFiles(session, List.of(tombstone,
emptyValue).iterator());
+ verify(readerFactory).createRecordReader(any(), any(), anyLong(),
any());
+ verify(session).transfer(flowFile, ConsumeKafka.PARSE_FAILURE);
+ verify(session, times(2)).adjustCounter(eq("Records Received from " +
FIRST_TOPIC), eq(1L), eq(false));
+ verify(session).adjustCounter(eq("Tombstones Skipped for " +
FIRST_TOPIC), eq(1L), eq(false));
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));
+ assertEquals(2L, offsetTracker.getRecordCounts().get(FIRST_TOPIC));
}
}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/WrapperRecordStreamKafkaMessageConverterTest.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/WrapperRecordStreamKafkaMessageConverterTest.java
index 4a7d83ecfa5..dd6b6520aca 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/WrapperRecordStreamKafkaMessageConverterTest.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/WrapperRecordStreamKafkaMessageConverterTest.java
@@ -88,12 +88,12 @@ class WrapperRecordStreamKafkaMessageConverterTest {
);
// Create ByteRecords
- final ByteRecord group1Record1 = new ByteRecord("topic1", 0, 0, 0L,
List.of(), null, "value1".getBytes(), 0L);
- final ByteRecord group1Record2 = new ByteRecord("topic1", 0, 3, 0L,
List.of(), null, "value4".getBytes(), 0L);
+ final ByteRecord group1Record1 = new ByteRecord("topic1", 0, 0, 0L,
List.of(), null, "value1".getBytes(), 0L, false);
+ final ByteRecord group1Record2 = new ByteRecord("topic1", 0, 3, 0L,
List.of(), null, "value4".getBytes(), 0L, false);
- final ByteRecord group2Record1 = new ByteRecord("topic1", 1, 1, 0L,
List.of(), null, "value2".getBytes(), 0L);
+ final ByteRecord group2Record1 = new ByteRecord("topic1", 1, 1, 0L,
List.of(), null, "value2".getBytes(), 0L, false);
- final ByteRecord group3Record1 = new ByteRecord("topic2", 0, 2, 0L,
List.of(), null, "value3".getBytes(), 0L);
+ final ByteRecord group3Record1 = new ByteRecord("topic2", 0, 2, 0L,
List.of(), null, "value3".getBytes(), 0L, false);
final Iterator<ByteRecord> consumerRecords = List.of(group1Record1,
group2Record1, group3Record1, group1Record2).iterator();
// Mock the session.create() and session.write() methods
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-api/src/main/java/org/apache/nifi/kafka/service/api/record/ByteRecord.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-api/src/main/java/org/apache/nifi/kafka/service/api/record/ByteRecord.java
index 4d57d661511..bb919e1fe6c 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-api/src/main/java/org/apache/nifi/kafka/service/api/record/ByteRecord.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-api/src/main/java/org/apache/nifi/kafka/service/api/record/ByteRecord.java
@@ -23,7 +23,7 @@ import java.util.Objects;
import java.util.Optional;
/**
- * Byte Record translation of Kafka Record with byte arrays for key and value
properties
+ * Byte Record translation of Kafka Record with byte arrays for key and value
properties.
*/
public class ByteRecord {
@@ -35,6 +35,7 @@ public class ByteRecord {
private final byte[] key;
private final byte[] value;
private final long bundledCount;
+ private final boolean tombstone;
public ByteRecord(
final String topic,
@@ -44,7 +45,8 @@ public class ByteRecord {
final List<RecordHeader> headers,
final byte[] key,
final byte[] value,
- final long bundledCount
+ final long bundledCount,
+ final boolean tombstone
) {
this.topic = Objects.requireNonNull(topic, "Topic required");
this.partition = partition;
@@ -54,6 +56,7 @@ public class ByteRecord {
this.key = key;
this.value = Objects.requireNonNull(value, "Value required");
this.bundledCount = bundledCount;
+ this.tombstone = tombstone;
}
public String getTopic() {
@@ -80,6 +83,10 @@ public class ByteRecord {
return value;
}
+ public boolean isTombstone() {
+ return tombstone;
+ }
+
public List<RecordHeader> getHeaders() {
return headers;
}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/consumer/Kafka3ConsumerService.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/consumer/Kafka3ConsumerService.java
index 37cb7e18254..ca388d4d868 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/consumer/Kafka3ConsumerService.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/consumer/Kafka3ConsumerService.java
@@ -375,11 +375,10 @@ public class Kafka3ConsumerService implements
KafkaConsumerService, Closeable, C
recordHeaders.add(recordHeader);
});
- // Support Kafka tombstones
- byte[] value = consumerRecord.value();
- if (value == null) {
- value = new byte[0];
- }
+ // A null Kafka value is a tombstone. Keep that distinct from a
non-null empty value.
+ final byte[] originalValue = consumerRecord.value();
+ final boolean isTombstone = originalValue == null;
+ final byte[] value = isTombstone ? new byte[0] : originalValue;
return new ByteRecord(
consumerRecord.topic(),
@@ -389,7 +388,8 @@ public class Kafka3ConsumerService implements
KafkaConsumerService, Closeable, C
recordHeaders,
consumerRecord.key(),
value,
- 1
+ 1,
+ isTombstone
);
}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/consumer/RecordIterable.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/consumer/RecordIterable.java
index 6f8229eb2f8..181221a74d4 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/consumer/RecordIterable.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/consumer/RecordIterable.java
@@ -58,11 +58,10 @@ public class RecordIterable implements Iterable<ByteRecord>
{
recordHeaders.add(recordHeader);
});
- // Support Kafka tombstones
- byte[] value = consumerRecord.value();
- if (value == null) {
- value = new byte[0];
- }
+ // A null Kafka value is a tombstone. Keep that distinct from a
non-null empty value.
+ final byte[] originalValue = consumerRecord.value();
+ final boolean isTombstone = originalValue == null;
+ final byte[] value = isTombstone ? new byte[0] : originalValue;
return new ByteRecord(
consumerRecord.topic(),
@@ -72,7 +71,8 @@ public class RecordIterable implements Iterable<ByteRecord> {
recordHeaders,
consumerRecord.key(),
value,
- 1
+ 1,
+ isTombstone
);
}
}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/consumer/Kafka3ConsumerServiceTest.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/consumer/Kafka3ConsumerServiceTest.java
index 0f59500515b..62dcdb8af4b 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/consumer/Kafka3ConsumerServiceTest.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/consumer/Kafka3ConsumerServiceTest.java
@@ -35,6 +35,7 @@ import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import java.time.Duration;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -249,7 +250,31 @@ class Kafka3ConsumerServiceTest {
verify(consumer, never()).commitSync(anyMap());
}
+ @Test
+ void testPollDistinguishesTombstoneFromEmptyValue() {
+ final TopicPartition partition0 = new TopicPartition(TOPIC,
PARTITION_0);
+ final ConsumerRecord<byte[], byte[]> tombstone = createRecord(TOPIC,
PARTITION_0, 0L, null);
+ final ConsumerRecord<byte[], byte[]> emptyValue = createRecord(TOPIC,
PARTITION_0, 1L, new byte[0]);
+
+ final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>>
recordsMap = new HashMap<>();
+ recordsMap.put(partition0, List.of(tombstone, emptyValue));
+
when(consumer.poll(any(Duration.class))).thenReturn(createConsumerRecords(recordsMap));
+
+ final List<ByteRecord> records = new ArrayList<>();
+ consumerService.poll(Duration.ofMillis(100)).forEach(records::add);
+
+ assertEquals(2, records.size());
+ assertTrue(records.get(0).isTombstone());
+ assertEquals(0, records.get(0).getValue().length);
+ assertFalse(records.get(1).isTombstone());
+ assertEquals(0, records.get(1).getValue().length);
+ }
+
private ConsumerRecord<byte[], byte[]> createRecord(final String topic,
final int partition, final long offset) {
+ return createRecord(topic, partition, offset, "test-value".getBytes());
+ }
+
+ private ConsumerRecord<byte[], byte[]> createRecord(final String topic,
final int partition, final long offset, final byte[] value) {
return new ConsumerRecord<>(
topic,
partition,
@@ -259,7 +284,7 @@ class Kafka3ConsumerServiceTest {
0,
0,
null,
- "test-value".getBytes(),
+ value,
new RecordHeaders(),
Optional.empty()
);
