diff --git a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-nar/pom.xml b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-nar/pom.xml
index ba809f5ee953..afaac17f94d5 100644
--- a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-nar/pom.xml
+++ b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-nar/pom.xml
@@ -31,6 +31,11 @@
nifi-kafka-processors
2.12.0-SNAPSHOT
+
+ org.apache.nifi
+ nifi-kafka-provenance-reporting-task
+ 2.12.0-SNAPSHOT
+
org.apache.nifi
nifi-kafka-service-api-nar
diff --git a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-provenance-reporting-task/pom.xml b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-provenance-reporting-task/pom.xml
new file mode 100644
index 000000000000..033d55ab0284
--- /dev/null
+++ b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-provenance-reporting-task/pom.xml
@@ -0,0 +1,65 @@
+
+
+
+ 4.0.0
+
+ org.apache.nifi
+ nifi-kafka-bundle
+ 2.12.0-SNAPSHOT
+
+ nifi-kafka-provenance-reporting-task
+ jar
+
+
+ org.apache.nifi
+ nifi-kafka-service-api
+ 2.12.0-SNAPSHOT
+ provided
+
+
+ org.apache.nifi
+ nifi-reporting-utils
+ 2.12.0-SNAPSHOT
+
+
+ org.apache.nifi
+ nifi-utils
+
+
+ org.apache.nifi
+ nifi-record-serialization-service-api
+ provided
+
+
+ org.apache.nifi
+ nifi-record
+ provided
+
+
+ org.apache.nifi
+ nifi-mock
+ test
+
+
+ org.apache.nifi
+ nifi-data-provenance-utils
+ 2.12.0-SNAPSHOT
+ test
+
+
+
diff --git a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-provenance-reporting-task/src/main/java/org/apache/nifi/kafka/reporting/KafkaProvenanceReportingTask.java b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-provenance-reporting-task/src/main/java/org/apache/nifi/kafka/reporting/KafkaProvenanceReportingTask.java
new file mode 100644
index 000000000000..032afe3a0eba
--- /dev/null
+++ b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-provenance-reporting-task/src/main/java/org/apache/nifi/kafka/reporting/KafkaProvenanceReportingTask.java
@@ -0,0 +1,1066 @@
+/*
+ * 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.reporting;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.controller.status.ProcessGroupStatus;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.kafka.service.api.KafkaConnectionService;
+import org.apache.nifi.kafka.service.api.producer.KafkaProducerService;
+import org.apache.nifi.kafka.service.api.producer.ProducerConfiguration;
+import org.apache.nifi.kafka.service.api.producer.PublishContext;
+import org.apache.nifi.kafka.service.api.record.KafkaRecord;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.provenance.ProvenanceEventRecord;
+import org.apache.nifi.provenance.ProvenanceEventType;
+import org.apache.nifi.reporting.AbstractReportingTask;
+import org.apache.nifi.reporting.ReportingContext;
+import org.apache.nifi.reporting.util.provenance.ComponentMapHolder;
+import org.apache.nifi.reporting.util.provenance.ProvenanceEventConsumer;
+import org.apache.nifi.schema.access.SchemaNotFoundException;
+import org.apache.nifi.serialization.RecordSetWriter;
+import org.apache.nifi.serialization.RecordSetWriterFactory;
+import org.apache.nifi.serialization.SimpleRecordSchema;
+import org.apache.nifi.serialization.record.MapRecord;
+import org.apache.nifi.serialization.record.RecordField;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.RecordSchema;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
+
+@Tags({"provenance", "lineage", "tracking", "kafka", "publish", "streaming", "avro", "record"})
+@CapabilityDescription(
+ "Publishes NiFi Provenance events directly to a Kafka topic using the Kafka Connection Service.")
+@Stateful(
+ scopes = Scope.LOCAL,
+ description = "Stores the ID of the last Provenance Event published to Kafka so that " +
+ "the task resumes from the correct position after a restart."
+)
+public class KafkaProvenanceReportingTask extends AbstractReportingTask {
+
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ static final String TIMESTAMP_FORMAT = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'";
+
+ static final DateTimeFormatter DATE_TIME_FORMATTER =
+ DateTimeFormatter.ofPattern(TIMESTAMP_FORMAT).withZone(ZoneOffset.UTC);
+
+ private static final String NIFI_API_PATH = "/nifi";
+
+ private static final String DEFAULT_PLATFORM = "nifi";
+
+ private static final String BOOLEAN_TRUE = "true";
+ private static final String BOOLEAN_FALSE = "false";
+
+ static final int NIFI_DEFAULT_MAX_ATTR_LENGTH = 65536;
+
+ static final int ATTR_LENGTH_MIN = 36;
+
+ private static final String FIELD_EVENT_ID = "eventId";
+ private static final String FIELD_EVENT_ORDINAL = "eventOrdinal";
+ private static final String FIELD_EVENT_TYPE = "eventType";
+ private static final String FIELD_TIMESTAMP_MILLIS = "timestampMillis";
+
+ private static final String FIELD_TIMESTAMP = "timestamp";
+ private static final String FIELD_DURATION_MILLIS = "durationMillis";
+ private static final String FIELD_LINEAGE_START = "lineageStart";
+ private static final String FIELD_DETAILS = "details";
+ private static final String FIELD_COMPONENT_ID = "componentId";
+ private static final String FIELD_COMPONENT_TYPE = "componentType";
+ private static final String FIELD_COMPONENT_NAME = "componentName";
+ private static final String FIELD_PROCESS_GROUP_ID = "processGroupId";
+ private static final String FIELD_PROCESS_GROUP_NAME = "processGroupName";
+ private static final String FIELD_ENTITY_ID = "entityId";
+ private static final String FIELD_ENTITY_TYPE = "entityType";
+ private static final String FIELD_ENTITY_SIZE = "entitySize";
+ private static final String FIELD_PREV_ENTITY_SIZE = "previousEntitySize";
+ private static final String FIELD_UPDATED_ATTRIBUTES = "updatedAttributes";
+ private static final String FIELD_PREV_ATTRIBUTES = "previousAttributes";
+ private static final String FIELD_ACTOR_HOSTNAME = "actorHostname";
+ private static final String FIELD_CONTENT_URI = "contentURI";
+ private static final String FIELD_PREV_CONTENT_URI = "previousContentURI";
+ private static final String FIELD_PARENT_IDS = "parentIds";
+ private static final String FIELD_CHILD_IDS = "childIds";
+ private static final String FIELD_PLATFORM = "platform";
+ private static final String FIELD_APPLICATION = "application";
+ private static final String FIELD_REMOTE_IDENTIFIER = "remoteIdentifier";
+ private static final String FIELD_ALT_IDENTIFIER = "alternateIdentifier";
+ private static final String FIELD_TRANSIT_URI = "transitUri";
+
+ private static final RecordSchema PROVENANCE_SCHEMA = buildProvenanceSchema();
+
+ private static RecordSchema buildProvenanceSchema() {
+ final List fields = new ArrayList<>();
+ fields.add(new RecordField(FIELD_EVENT_ID, RecordFieldType.STRING.getDataType()));
+ fields.add(new RecordField(FIELD_EVENT_ORDINAL, RecordFieldType.LONG.getDataType()));
+ fields.add(new RecordField(FIELD_EVENT_TYPE, RecordFieldType.STRING.getDataType()));
+ fields.add(new RecordField(FIELD_TIMESTAMP_MILLIS, RecordFieldType.LONG.getDataType()));
+ fields.add(new RecordField(FIELD_DURATION_MILLIS, RecordFieldType.LONG.getDataType()));
+ fields.add(new RecordField(FIELD_LINEAGE_START, RecordFieldType.LONG.getDataType()));
+ fields.add(new RecordField(FIELD_DETAILS, RecordFieldType.STRING.getDataType()));
+ addStringFields(fields, FIELD_COMPONENT_ID, FIELD_COMPONENT_TYPE, FIELD_COMPONENT_NAME,
+ FIELD_PROCESS_GROUP_ID, FIELD_PROCESS_GROUP_NAME, FIELD_ENTITY_ID, FIELD_ENTITY_TYPE);
+ fields.add(new RecordField(FIELD_ENTITY_SIZE, RecordFieldType.LONG.getDataType()));
+ fields.add(new RecordField(FIELD_PREV_ENTITY_SIZE, RecordFieldType.LONG.getDataType()));
+ fields.add(new RecordField(FIELD_UPDATED_ATTRIBUTES,
+ RecordFieldType.MAP.getMapDataType(RecordFieldType.STRING.getDataType())));
+ fields.add(new RecordField(FIELD_PREV_ATTRIBUTES,
+ RecordFieldType.MAP.getMapDataType(RecordFieldType.STRING.getDataType())));
+ addStringFields(fields, FIELD_ACTOR_HOSTNAME, FIELD_CONTENT_URI, FIELD_PREV_CONTENT_URI);
+ fields.add(new RecordField(FIELD_PARENT_IDS,
+ RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.STRING.getDataType())));
+ fields.add(new RecordField(FIELD_CHILD_IDS,
+ RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.STRING.getDataType())));
+ addStringFields(fields, FIELD_PLATFORM, FIELD_APPLICATION,
+ FIELD_REMOTE_IDENTIFIER, FIELD_ALT_IDENTIFIER, FIELD_TRANSIT_URI);
+ return new SimpleRecordSchema(fields);
+ }
+
+ private static void addStringFields(final List fields, final String... names) {
+ for (final String name : names) {
+ fields.add(new RecordField(name, RecordFieldType.STRING.getDataType()));
+ }
+ }
+
+ static final AllowableValue BEGINNING_OF_STREAM = new AllowableValue(
+ "beginning-of-stream", "Beginning of Stream",
+ "Start reading from the oldest event in the stream.");
+ static final AllowableValue END_OF_STREAM = new AllowableValue(
+ "end-of-stream", "End of Stream",
+ "Start reading from the current end of the stream, ignoring historical events.");
+
+ static final AllowableValue KEY_NONE = new AllowableValue(
+ "none", "None",
+ "No key is set on Kafka messages. Kafka assigns messages to partitions using its default strategy.");
+ static final AllowableValue KEY_LINEAGE_START = new AllowableValue(
+ FIELD_LINEAGE_START, "Lineage Start",
+ "Uses the lineage start timestamp (epoch millis) as the key. " +
+ "All events belonging to the same lineage chain share the same key and land on the same partition, " +
+ "enabling ordered consumption of a complete data lineage.");
+ static final AllowableValue KEY_ENTITY_ID = new AllowableValue(
+ FIELD_ENTITY_ID, "FlowFile UUID",
+ "Uses the FlowFile UUID as the key. All provenance events for the same FlowFile share the same key.");
+ static final AllowableValue KEY_COMPONENT_ID = new AllowableValue(
+ FIELD_COMPONENT_ID, "Component ID",
+ "Uses the component ID (processor/connection UUID) as the key. Groups all events from the same component.");
+ static final AllowableValue KEY_EVENT_TYPE = new AllowableValue(
+ FIELD_EVENT_TYPE, "Event Type",
+ "Uses the event type name (e.g. CREATE, SEND, RECEIVE) as the key. Useful for per-type topic compaction.");
+ static final AllowableValue KEY_EVENT_ORDINAL = new AllowableValue(
+ FIELD_EVENT_ORDINAL, "Event Ordinal",
+ "Uses the event's unique sequential ordinal ID (long) as the key. Each message has a distinct key.");
+
+ static final AllowableValue DELIVERY_REPLICATED = new AllowableValue(
+ "all", "Guarantee Replicated Delivery",
+ "Producer waits for acknowledgment from all in-sync replicas. Strongest guarantee.");
+ static final AllowableValue DELIVERY_ONE_NODE = new AllowableValue(
+ "1", "Guarantee Single Node Delivery",
+ "Producer waits for acknowledgment from the partition leader only.");
+ static final AllowableValue DELIVERY_BEST_EFFORT = new AllowableValue(
+ "0", "Best Effort",
+ "No acknowledgment required. Highest throughput, possible data loss on broker failure.");
+
+ static final PropertyDescriptor KAFKA_CONNECTION_SERVICE = new PropertyDescriptor.Builder()
+ .name("kafka-connection-service")
+ .displayName("Kafka Connection Service")
+ .description("The Kafka Connection Service to use for connecting to Kafka brokers.")
+ .identifiesControllerService(KafkaConnectionService.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor TOPIC_NAME = new PropertyDescriptor.Builder()
+ .name("topic-name")
+ .displayName("Topic Name")
+ .description("The Kafka topic to which Provenance Events are published.")
+ .required(true)
+ .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .build();
+
+ static final PropertyDescriptor RECORD_WRITER = new PropertyDescriptor.Builder()
+ .name("record-writer")
+ .displayName("Record Writer")
+ .description(
+ "Specifies the Controller Service to use for serializing Provenance Events before publishing to Kafka. " +
+ "When set, all events in a batch are encoded as a record set using the configured writer " +
+ "(e.g. AvroRecordSetWriter, JsonRecordSetWriter) and published as a single Kafka message per batch. " +
+ "Configure AvroRecordSetWriter with a Schema Registry and 'Schema Write Strategy = Confluent encoded' " +
+ "to emit the standard Confluent wire format (magic byte 0x00 + 4-byte schema ID + Avro binary), " +
+ "which allows consumers to resolve the schema by ID from the registry. " +
+ "The record schema is identical to the one used by SiteToSiteProvenanceReportingTask. " +
+ "When not set, each event is serialized as a JSON object.")
+ .identifiesControllerService(RecordSetWriterFactory.class)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor MESSAGE_KEY_FIELD = new PropertyDescriptor.Builder()
+ .name("message-key-field")
+ .displayName("Message Key Field")
+ .description(
+ "Specifies which Provenance Event field to use as the Kafka message key. " +
+ "A meaningful key enables Kafka to co-locate related events on the same partition, " +
+ "which is important for ordered consumption and log compaction. " +
+ "The key value is serialized as a UTF-8 string.")
+ .required(true)
+ .allowableValues(KEY_NONE, KEY_LINEAGE_START, KEY_ENTITY_ID,
+ KEY_COMPONENT_ID, KEY_EVENT_TYPE, KEY_EVENT_ORDINAL)
+ .defaultValue(KEY_LINEAGE_START.getValue())
+ .build();
+
+ static final PropertyDescriptor DELIVERY_GUARANTEE = new PropertyDescriptor.Builder()
+ .name("delivery-guarantee")
+ .displayName("Delivery Guarantee")
+ .description("Level of delivery guarantee required (maps to Kafka producer 'acks').")
+ .required(true)
+ .allowableValues(DELIVERY_REPLICATED, DELIVERY_ONE_NODE, DELIVERY_BEST_EFFORT)
+ .defaultValue(DELIVERY_REPLICATED.getValue())
+ .build();
+
+ static final PropertyDescriptor COMPRESSION_TYPE = new PropertyDescriptor.Builder()
+ .name("compression-type")
+ .displayName("Compression Type")
+ .description("Compression codec for Kafka producer batches.")
+ .required(true)
+ .allowableValues("none", "gzip", "snappy", "lz4", "zstd")
+ .defaultValue("none")
+ .build();
+
+ static final PropertyDescriptor MAX_REQUEST_SIZE = new PropertyDescriptor.Builder()
+ .name("max-request-size")
+ .displayName("Max Request Size")
+ .description("Maximum Kafka producer request size (maps to 'max.request.size').")
+ .required(true)
+ .defaultValue("1 MB")
+ .addValidator(StandardValidators.DATA_SIZE_VALIDATOR)
+ .build();
+
+ static final PropertyDescriptor TRANSACTIONS_ENABLED = new PropertyDescriptor.Builder()
+ .name("transactions-enabled")
+ .displayName("Transactions Enabled")
+ .description("Whether to use Kafka transactions. Each batch is published atomically when enabled.")
+ .required(true)
+ .allowableValues(BOOLEAN_TRUE, BOOLEAN_FALSE)
+ .defaultValue(BOOLEAN_TRUE)
+ .build();
+
+ static final PropertyDescriptor TRANSACTIONAL_ID_PREFIX = new PropertyDescriptor.Builder()
+ .name("transactional-id-prefix")
+ .displayName("Transactional ID Prefix")
+ .description("Prefix for the auto-generated transactional.id (used when Transactions Enabled is true).")
+ .required(false)
+ .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .defaultValue("nifi-provenance-")
+ .dependsOn(TRANSACTIONS_ENABLED, BOOLEAN_TRUE)
+ .build();
+
+ static final PropertyDescriptor INSTANCE_URL = new PropertyDescriptor.Builder()
+ .name("instance-url")
+ .displayName("Instance URL")
+ .description("URL of this NiFi instance (ending with /nifi). Used to generate contentURI fields. Optional.")
+ .required(false)
+ .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
+ .defaultValue("https://${hostname(true)}:8443/nifi")
+ .addValidator(StandardValidators.URL_VALIDATOR)
+ .build();
+
+ static final PropertyDescriptor PLATFORM = new PropertyDescriptor.Builder()
+ .name(FIELD_PLATFORM)
+ .displayName("Platform")
+ .description("Environment label embedded in each record (e.g. 'prod-nifi-cluster-01'). Must be non-empty.")
+ .required(true)
+ .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
+ .defaultValue(DEFAULT_PLATFORM)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("batch-size")
+ .displayName("Batch Size")
+ .description("Maximum number of Provenance Events to collect per scheduling trigger. " +
+ "Each event is published as an individual Kafka message; all messages from one batch are sent in a single producer call.")
+ .required(true)
+ .defaultValue("1000")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .build();
+
+ static final PropertyDescriptor ALLOW_NULL_VALUES = new PropertyDescriptor.Builder()
+ .name("allow-null-values")
+ .displayName("Include Null Values")
+ .description(
+ "When true, null-valued fields are included in the JSON output. " +
+ "Has no effect when Record Writer is configured.")
+ .required(true)
+ .allowableValues(BOOLEAN_TRUE, BOOLEAN_FALSE)
+ .defaultValue(BOOLEAN_FALSE)
+ .build();
+
+ static final PropertyDescriptor START_POSITION = new PropertyDescriptor.Builder()
+ .name("start-position")
+ .displayName("Start Position")
+ .description("Where to begin reading if no prior state exists.")
+ .allowableValues(BEGINNING_OF_STREAM, END_OF_STREAM)
+ .defaultValue(BEGINNING_OF_STREAM.getValue())
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor FILTER_EVENT_TYPE = new PropertyDescriptor.Builder()
+ .name("Event Type to Include")
+ .displayName("Event Type to Include")
+ .description("Comma-separated list of ProvenanceEventType values to include. Available: "
+ + Arrays.deepToString(ProvenanceEventType.values()))
+ .required(false)
+ .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ static final PropertyDescriptor FILTER_EVENT_TYPE_EXCLUDE = new PropertyDescriptor.Builder()
+ .name("Event Type to Exclude")
+ .displayName("Event Type to Exclude")
+ .description("Comma-separated list of ProvenanceEventType values to exclude (takes precedence over include).")
+ .required(false)
+ .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ static final PropertyDescriptor FILTER_COMPONENT_TYPE = new PropertyDescriptor.Builder()
+ .name("Component Type to Include")
+ .displayName("Component Type to Include")
+ .description("Regex to include events by component type.")
+ .required(false)
+ .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
+ .addValidator(StandardValidators.REGULAR_EXPRESSION_VALIDATOR)
+ .build();
+
+ static final PropertyDescriptor FILTER_COMPONENT_TYPE_EXCLUDE = new PropertyDescriptor.Builder()
+ .name("Component Type to Exclude")
+ .displayName("Component Type to Exclude")
+ .description("Regex to exclude events by component type (takes precedence over include).")
+ .required(false)
+ .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
+ .addValidator(StandardValidators.REGULAR_EXPRESSION_VALIDATOR)
+ .build();
+
+ static final PropertyDescriptor FILTER_COMPONENT_ID = new PropertyDescriptor.Builder()
+ .name("Component ID to Include")
+ .displayName("Component ID to Include")
+ .description("Comma-separated processor/connection UUIDs to include.")
+ .required(false)
+ .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ static final PropertyDescriptor FILTER_COMPONENT_ID_EXCLUDE = new PropertyDescriptor.Builder()
+ .name("Component ID to Exclude")
+ .displayName("Component ID to Exclude")
+ .description("Comma-separated processor/connection UUIDs to exclude (takes precedence over include).")
+ .required(false)
+ .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ static final PropertyDescriptor FILTER_COMPONENT_NAME = new PropertyDescriptor.Builder()
+ .name("Component Name to Include")
+ .displayName("Component Name to Include")
+ .description("Regex to include events by component name.")
+ .required(false)
+ .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
+ .addValidator(StandardValidators.REGULAR_EXPRESSION_VALIDATOR)
+ .build();
+
+ static final PropertyDescriptor FILTER_COMPONENT_NAME_EXCLUDE = new PropertyDescriptor.Builder()
+ .name("Component Name to Exclude")
+ .displayName("Component Name to Exclude")
+ .description("Regex to exclude events by component name (takes precedence over include).")
+ .required(false)
+ .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
+ .addValidator(StandardValidators.REGULAR_EXPRESSION_VALIDATOR)
+ .build();
+
+ static final PropertyDescriptor FILTER_ATTRIBUTES_INCLUDE = new PropertyDescriptor.Builder()
+ .name("Attributes to Include")
+ .displayName("Attributes to Include")
+ .description("Comma-separated list of attribute names or regular expressions. " +
+ "When set, only attributes whose names match any entry are included in " +
+ "previousAttributes and updatedAttributes. " +
+ "Cannot be configured together with attributes to exclude.")
+ .required(false)
+ .addValidator(createAttributeFilterValidator())
+ .build();
+
+ static final PropertyDescriptor FILTER_ATTRIBUTES_EXCLUDE = new PropertyDescriptor.Builder()
+ .name("Attributes to Exclude")
+ .displayName("Attributes to Exclude")
+ .description("Comma-separated list of attribute names or regular expressions. " +
+ "When set, attributes whose names match any entry are excluded from " +
+ "previousAttributes and updatedAttributes; all others are included. " +
+ "Cannot be configured together with attributes to include.")
+ .required(false)
+ .addValidator(createAttributeFilterValidator())
+ .build();
+
+ static final PropertyDescriptor ATTRIBUTE_MAX_LENGTH = new PropertyDescriptor.Builder()
+ .name("attribute-max-length")
+ .displayName("Attribute Max Length")
+ .description("Maximum number of characters to include in each attribute value sent in Kafka provenance events. " +
+ "Values exceeding this limit are trimmed. " +
+ "Must be at least " + ATTR_LENGTH_MIN + " to preserve UUID attributes. " +
+ "If set to " + NIFI_DEFAULT_MAX_ATTR_LENGTH + " or higher, trimming is not applied because " +
+ "NiFi's global provenance repository limit already enforces that ceiling. " +
+ "Note: this task compares against the NiFi default global limit of " + NIFI_DEFAULT_MAX_ATTR_LENGTH + " characters. " +
+ "If the NiFi property nifi.provenance.repository.max.attribute.length has been set to a lower value by an administrator, " +
+ "attribute values will already be capped at that lower global limit before they reach this task. " +
+ "In that case, setting Attribute Max Length to a value larger than the actual global limit has no trimming effect, " +
+ "even if it is below " + NIFI_DEFAULT_MAX_ATTR_LENGTH + ", " +
+ "and no warning will be logged because this task cannot read the runtime value of nifi.provenance.repository.max.attribute.length. " +
+ "To ensure trimming takes effect, set Attribute Max Length to a value strictly smaller than " +
+ "the configured nifi.provenance.repository.max.attribute.length.")
+ .required(false)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .build();
+
+ private static final List PROPERTY_DESCRIPTORS = List.of(
+ KAFKA_CONNECTION_SERVICE,
+ TOPIC_NAME,
+ RECORD_WRITER,
+ MESSAGE_KEY_FIELD,
+ DELIVERY_GUARANTEE,
+ COMPRESSION_TYPE,
+ MAX_REQUEST_SIZE,
+ TRANSACTIONS_ENABLED,
+ TRANSACTIONAL_ID_PREFIX,
+ INSTANCE_URL,
+ PLATFORM,
+ BATCH_SIZE,
+ ALLOW_NULL_VALUES,
+ START_POSITION,
+ FILTER_EVENT_TYPE,
+ FILTER_EVENT_TYPE_EXCLUDE,
+ FILTER_COMPONENT_TYPE,
+ FILTER_COMPONENT_TYPE_EXCLUDE,
+ FILTER_COMPONENT_ID,
+ FILTER_COMPONENT_ID_EXCLUDE,
+ FILTER_COMPONENT_NAME,
+ FILTER_COMPONENT_NAME_EXCLUDE,
+ FILTER_ATTRIBUTES_INCLUDE,
+ FILTER_ATTRIBUTES_EXCLUDE,
+ ATTRIBUTE_MAX_LENGTH
+ );
+
+ private record EncodingContext(
+ String hostname,
+ String nifiUrlBase,
+ String applicationName,
+ String platform,
+ String nodeIdentifier,
+ boolean allowNullValues) {
+ }
+
+ private record KafkaOutputConfig(
+ String topicName,
+ String messageKeyField,
+ RecordSetWriterFactory writerFactory) {
+ }
+
+ private final AtomicReference consumerRef = new AtomicReference<>();
+ private final AtomicReference producerRef = new AtomicReference<>();
+
+ final AtomicReference> attributeIncludePatterns = new AtomicReference<>(Collections.emptyList());
+ final AtomicReference> attributeExcludePatterns = new AtomicReference<>(Collections.emptyList());
+ volatile int attributeMaxLength = -1;
+
+ @Override
+ protected List getSupportedPropertyDescriptors() {
+ return PROPERTY_DESCRIPTORS;
+ }
+
+ @Override
+ protected Collection customValidate(final ValidationContext context) {
+ final List results = new ArrayList<>();
+ final boolean hasIncludeFilter = context.getProperty(FILTER_ATTRIBUTES_INCLUDE).isSet();
+ final boolean hasExcludeFilter = context.getProperty(FILTER_ATTRIBUTES_EXCLUDE).isSet();
+ if (hasIncludeFilter && hasExcludeFilter) {
+ results.add(new ValidationResult.Builder()
+ .subject(FILTER_ATTRIBUTES_INCLUDE.getDisplayName() + " / " + FILTER_ATTRIBUTES_EXCLUDE.getDisplayName())
+ .valid(false)
+ .explanation("'Attributes to Include' and 'Attributes to Exclude' are mutually exclusive. Configure only one.")
+ .build());
+ }
+ return results;
+ }
+
+ /**
+ * Returns a {@link Validator} that accepts a comma-separated list of attribute names or regular
+ * expressions. Each entry is compiled as a {@link Pattern} to verify it is syntactically valid.
+ * Literal attribute names are also valid regular expressions, so this validator covers both cases.
+ */
+ private static Validator createAttributeFilterValidator() {
+ return (subject, value, context) -> {
+ if (value == null || value.isBlank()) {
+ return new ValidationResult.Builder().subject(subject).input(value).valid(true).build();
+ }
+ for (final String entry : value.split(",")) {
+ final String trimmed = entry.trim();
+ if (trimmed.isEmpty()) {
+ continue;
+ }
+ try {
+ Pattern.compile(trimmed);
+ } catch (final PatternSyntaxException e) {
+ return new ValidationResult.Builder()
+ .subject(subject)
+ .input(value)
+ .valid(false)
+ .explanation("'" + trimmed + "' is not a valid regular expression: " + e.getMessage())
+ .build();
+ }
+ }
+ return new ValidationResult.Builder().subject(subject).input(value).valid(true).build();
+ };
+ }
+
+ @OnScheduled
+ public void onScheduled(final ConfigurationContext context) {
+ final KafkaConnectionService connectionService =
+ context.getProperty(KAFKA_CONNECTION_SERVICE).asControllerService(KafkaConnectionService.class);
+ producerRef.set(connectionService.getProducerService(buildProducerConfiguration(context)));
+
+ final ProvenanceEventConsumer consumer = new ProvenanceEventConsumer();
+ consumer.setStartPositionValue(context.getProperty(START_POSITION).getValue());
+ consumer.setBatchSize(context.getProperty(BATCH_SIZE).asInteger());
+ consumer.setLogger(getLogger());
+
+ consumer.setComponentTypeRegex(
+ context.getProperty(FILTER_COMPONENT_TYPE).evaluateAttributeExpressions().getValue());
+ consumer.setComponentTypeRegexExclude(
+ context.getProperty(FILTER_COMPONENT_TYPE_EXCLUDE).evaluateAttributeExpressions().getValue());
+ consumer.setComponentNameRegex(
+ context.getProperty(FILTER_COMPONENT_NAME).evaluateAttributeExpressions().getValue());
+ consumer.setComponentNameRegexExclude(
+ context.getProperty(FILTER_COMPONENT_NAME_EXCLUDE).evaluateAttributeExpressions().getValue());
+
+ final String[] includeEventTypes = StringUtils.stripAll(StringUtils.split(
+ context.getProperty(FILTER_EVENT_TYPE).evaluateAttributeExpressions().getValue(), ','));
+ if (includeEventTypes != null) {
+ for (final String type : includeEventTypes) {
+ try {
+ consumer.addTargetEventType(ProvenanceEventType.valueOf(type));
+ } catch (final IllegalArgumentException e) {
+ getLogger().warn("'{}' is not a valid ProvenanceEventType; ignored from include filter.", type);
+ }
+ }
+ }
+
+ final String[] excludeEventTypes = StringUtils.stripAll(StringUtils.split(
+ context.getProperty(FILTER_EVENT_TYPE_EXCLUDE).evaluateAttributeExpressions().getValue(), ','));
+ if (excludeEventTypes != null) {
+ for (final String type : excludeEventTypes) {
+ try {
+ consumer.addTargetEventTypeExclude(ProvenanceEventType.valueOf(type));
+ } catch (final IllegalArgumentException e) {
+ getLogger().warn("'{}' is not a valid ProvenanceEventType; ignored from exclude filter.", type);
+ }
+ }
+ }
+
+ final String[] includeComponentIds = StringUtils.stripAll(StringUtils.split(
+ context.getProperty(FILTER_COMPONENT_ID).evaluateAttributeExpressions().getValue(), ','));
+ if (includeComponentIds != null) {
+ consumer.addTargetComponentId(includeComponentIds);
+ }
+
+ final String[] excludeComponentIds = StringUtils.stripAll(StringUtils.split(
+ context.getProperty(FILTER_COMPONENT_ID_EXCLUDE).evaluateAttributeExpressions().getValue(), ','));
+ if (excludeComponentIds != null) {
+ consumer.addTargetComponentIdExclude(excludeComponentIds);
+ }
+
+ attributeIncludePatterns.set(parsePatterns(context.getProperty(FILTER_ATTRIBUTES_INCLUDE).getValue()));
+ attributeExcludePatterns.set(parsePatterns(context.getProperty(FILTER_ATTRIBUTES_EXCLUDE).getValue()));
+ attributeMaxLength = resolveAttributeMaxLength(context.getProperty(ATTRIBUTE_MAX_LENGTH).asInteger());
+
+ consumer.setScheduled(true);
+ consumerRef.set(consumer);
+ }
+
+ @OnUnscheduled
+ public void onUnscheduled() {
+ final ProvenanceEventConsumer consumer = consumerRef.get();
+ if (consumer != null) {
+ consumer.setScheduled(false);
+ }
+ }
+
+ @OnStopped
+ public void onStopped() {
+ final KafkaProducerService producer = producerRef.getAndSet(null);
+ if (producer != null) {
+ try {
+ producer.close();
+ } catch (final Exception e) {
+ getLogger().warn("Failed to close KafkaProducerService cleanly", e);
+ }
+ }
+ }
+
+ @Override
+ public void onTrigger(final ReportingContext context) {
+ final boolean isClustered = context.isClustered();
+ final String nodeId = context.getClusterNodeIdentifier();
+ if (nodeId == null && isClustered) {
+ getLogger().debug("Cluster Node Identifier not yet established; skipping trigger.");
+ return;
+ }
+
+ final KafkaProducerService producer = producerRef.get();
+ if (producer == null || producer.isClosed()) {
+ getLogger().warn("KafkaProducerService is not available; skipping trigger.");
+ return;
+ }
+
+ final ProcessGroupStatus rootStatus = context.getEventAccess().getControllerStatus();
+ final String topicName = context.getProperty(TOPIC_NAME).evaluateAttributeExpressions().getValue();
+ final String messageKeyField = context.getProperty(MESSAGE_KEY_FIELD).getValue();
+ final boolean useRecordWriter = context.getProperty(RECORD_WRITER).isSet();
+ final RecordSetWriterFactory writerFactory = useRecordWriter
+ ? context.getProperty(RECORD_WRITER).asControllerService(RecordSetWriterFactory.class) : null;
+
+ final EncodingContext encodingCtx = buildEncodingContext(context, rootStatus, nodeId);
+ final KafkaOutputConfig kafkaConfig = new KafkaOutputConfig(topicName, messageKeyField, writerFactory);
+
+ try {
+ consumerRef.get().consumeEvents(context, (mapHolder, events) ->
+ publishBatch(mapHolder, events, producer, kafkaConfig, encodingCtx));
+ } catch (final ProcessException pe) {
+ getLogger().error("Failed to publish Provenance Events to Kafka topic '{}'", topicName, pe);
+ }
+ }
+
+ private void publishBatch(
+ final ComponentMapHolder mapHolder,
+ final List events,
+ final KafkaProducerService producer,
+ final KafkaOutputConfig kafkaConfig,
+ final EncodingContext encodingCtx) {
+
+ final List kafkaRecords = new ArrayList<>(events.size());
+ for (final ProvenanceEventRecord event : events) {
+ final String componentName = mapHolder.getComponentName(event.getComponentId());
+ final String processGroupId = mapHolder.getProcessGroupId(event.getComponentId(), event.getComponentType());
+ final String processGroupName = mapHolder.getComponentName(processGroupId);
+ try {
+ kafkaRecords.add(toKafkaRecord(event, componentName, processGroupId, processGroupName,
+ kafkaConfig, encodingCtx));
+ } catch (final IOException e) {
+ throw new ProcessException("Failed to encode Provenance Event", e);
+ }
+ }
+
+ final PublishContext publishContext = new PublishContext(kafkaConfig.topicName(), null, null, null);
+ producer.send(kafkaRecords.iterator(), publishContext);
+ producer.complete();
+
+ if (publishContext.getException() != null) {
+ throw new ProcessException(
+ "Kafka producer reported an error after send", publishContext.getException());
+ }
+
+ getLogger().debug("Published {} Provenance Events to Kafka topic '{}' (format: {})",
+ events.size(), kafkaConfig.topicName(),
+ kafkaConfig.writerFactory() != null ? "record-writer" : "json");
+ }
+
+ private KafkaRecord toKafkaRecord(
+ final ProvenanceEventRecord event,
+ final String componentName,
+ final String processGroupId,
+ final String processGroupName,
+ final KafkaOutputConfig kafkaConfig,
+ final EncodingContext ctx) throws IOException {
+
+ final byte[] key = resolveMessageKey(event, kafkaConfig.messageKeyField());
+ final byte[] payload;
+ if (kafkaConfig.writerFactory() != null) {
+ final MapRecord mapRecord = buildProvenanceRecord(
+ event, componentName, processGroupId, processGroupName, ctx);
+ payload = encodeRecord(kafkaConfig.writerFactory(), mapRecord);
+ } else {
+ payload = serializeToJson(event, componentName, processGroupId, processGroupName, ctx)
+ .toString().getBytes(StandardCharsets.UTF_8);
+ }
+ return new KafkaRecord(kafkaConfig.topicName(), null, event.getEventTime(), key, payload,
+ Collections.emptyList());
+ }
+
+ private EncodingContext buildEncodingContext(
+ final ReportingContext context,
+ final ProcessGroupStatus rootStatus,
+ final String nodeId) {
+
+ final String nifiUrlString = context.getProperty(INSTANCE_URL).evaluateAttributeExpressions().getValue();
+ final String platform = context.getProperty(PLATFORM).evaluateAttributeExpressions().getValue();
+ final boolean allowNullValues = context.getProperty(ALLOW_NULL_VALUES).asBoolean();
+ final String rootGroupName = rootStatus == null ? null : rootStatus.getName();
+
+ String hostname = null;
+ String nifiUrlBase = null;
+ if (nifiUrlString != null && !nifiUrlString.isBlank()) {
+ try {
+ final URL nifiUrl = URI.create(nifiUrlString).toURL();
+ hostname = nifiUrl.getHost();
+ nifiUrlBase = resolveUrlBase(nifiUrl);
+ } catch (final IllegalArgumentException | MalformedURLException e) {
+ getLogger().warn("Configured Instance URL '{}' is invalid; contentURI fields will be omitted.",
+ nifiUrlString);
+ }
+ }
+
+ return new EncodingContext(hostname, nifiUrlBase, rootGroupName, platform, nodeId, allowNullValues);
+ }
+
+ private MapRecord buildProvenanceRecord(
+ final ProvenanceEventRecord event,
+ final String componentName,
+ final String processGroupId,
+ final String processGroupName,
+ final EncodingContext ctx) {
+
+ final String contentBase = resolveContentBase(event, ctx);
+
+ final Map values = new LinkedHashMap<>();
+ values.put(FIELD_EVENT_ID, UUID.randomUUID().toString());
+ values.put(FIELD_EVENT_ORDINAL, event.getEventId());
+ values.put(FIELD_EVENT_TYPE, event.getEventType().name());
+ values.put(FIELD_TIMESTAMP_MILLIS, event.getEventTime());
+ values.put(FIELD_DURATION_MILLIS, event.getEventDuration());
+ values.put(FIELD_LINEAGE_START, event.getLineageStartDate());
+ values.put(FIELD_DETAILS, event.getDetails());
+ values.put(FIELD_COMPONENT_ID, event.getComponentId());
+ values.put(FIELD_COMPONENT_TYPE, event.getComponentType());
+ values.put(FIELD_COMPONENT_NAME, componentName);
+ values.put(FIELD_PROCESS_GROUP_ID, processGroupId);
+ values.put(FIELD_PROCESS_GROUP_NAME, processGroupName);
+ values.put(FIELD_ENTITY_ID, event.getFlowFileUuid());
+ values.put(FIELD_ENTITY_TYPE, "org.apache.nifi.flowfile.FlowFile");
+ values.put(FIELD_ENTITY_SIZE, event.getFileSize());
+ values.put(FIELD_PREV_ENTITY_SIZE, event.getPreviousFileSize());
+ values.put(FIELD_UPDATED_ATTRIBUTES,
+ filterAttributes(event.getUpdatedAttributes() != null
+ ? event.getUpdatedAttributes() : Collections.emptyMap()));
+ values.put(FIELD_PREV_ATTRIBUTES,
+ filterAttributes(event.getPreviousAttributes() != null
+ ? event.getPreviousAttributes() : Collections.emptyMap()));
+ values.put(FIELD_ACTOR_HOSTNAME, ctx.hostname());
+ if (contentBase != null) {
+ final String clusterSuffix = resolveClusterSuffix(ctx);
+ values.put(FIELD_CONTENT_URI, contentBase + "output" + clusterSuffix);
+ values.put(FIELD_PREV_CONTENT_URI, contentBase + "input" + clusterSuffix);
+ } else {
+ values.put(FIELD_CONTENT_URI, null);
+ values.put(FIELD_PREV_CONTENT_URI, null);
+ }
+ values.put(FIELD_PARENT_IDS,
+ event.getParentUuids() != null ? new ArrayList<>(event.getParentUuids()) : Collections.emptyList());
+ values.put(FIELD_CHILD_IDS,
+ event.getChildUuids() != null ? new ArrayList<>(event.getChildUuids()) : Collections.emptyList());
+ values.put(FIELD_PLATFORM, ctx.platform());
+ values.put(FIELD_APPLICATION, ctx.applicationName() != null ? ctx.applicationName() : "");
+ values.put(FIELD_REMOTE_IDENTIFIER, event.getSourceSystemFlowFileIdentifier());
+ values.put(FIELD_ALT_IDENTIFIER, event.getAlternateIdentifierUri());
+ values.put(FIELD_TRANSIT_URI, event.getTransitUri());
+
+ return new MapRecord(PROVENANCE_SCHEMA, values);
+ }
+
+ private static List parsePatterns(final String value) {
+ if (value == null || value.isBlank()) {
+ return Collections.emptyList();
+ }
+ final List patterns = new ArrayList<>();
+ for (final String entry : value.split(",")) {
+ final String trimmed = entry.trim();
+ if (!trimmed.isEmpty()) {
+ patterns.add(Pattern.compile(trimmed));
+ }
+ }
+ return Collections.unmodifiableList(patterns);
+ }
+
+ private int resolveAttributeMaxLength(final Integer configured) {
+ if (configured == null) {
+ return -1;
+ }
+ int effective = configured;
+ if (effective < ATTR_LENGTH_MIN) {
+ getLogger().warn(
+ "Configured '{}' is {}; enforcing minimum of {} to preserve UUID attributes.",
+ ATTRIBUTE_MAX_LENGTH.getDisplayName(), effective, ATTR_LENGTH_MIN);
+ effective = ATTR_LENGTH_MIN;
+ }
+ if (effective >= NIFI_DEFAULT_MAX_ATTR_LENGTH) {
+ getLogger().info(
+ "Configured '{}' ({}) is greater than or equal to the NiFi global provenance limit ({}). " +
+ "Attribute value trimming will not be applied because NiFi already enforces that ceiling.",
+ ATTRIBUTE_MAX_LENGTH.getDisplayName(), effective, NIFI_DEFAULT_MAX_ATTR_LENGTH);
+ return -1;
+ }
+ return effective;
+ }
+
+ private Map filterAttributes(final Map attributes) {
+ if (attributes == null) {
+ return Collections.emptyMap();
+ }
+ final List includePatterns = attributeIncludePatterns.get();
+ final List excludePatterns = attributeExcludePatterns.get();
+ final int maxLength = attributeMaxLength;
+
+ if (includePatterns.isEmpty() && excludePatterns.isEmpty() && maxLength < 0) {
+ return attributes;
+ }
+
+ final Map filtered = new LinkedHashMap<>();
+ for (final Map.Entry entry : attributes.entrySet()) {
+ final String key = entry.getKey();
+ if (key == null
+ || (!includePatterns.isEmpty() && !matchesAny(key, includePatterns))
+ || (!excludePatterns.isEmpty() && matchesAny(key, excludePatterns))) {
+ continue;
+ }
+ String attrValue = entry.getValue();
+ if (attrValue != null && maxLength > 0 && attrValue.length() > maxLength) {
+ attrValue = attrValue.substring(0, maxLength);
+ }
+ filtered.put(key, attrValue);
+ }
+ return filtered;
+ }
+
+ private static boolean matchesAny(final String key, final List patterns) {
+ for (final Pattern pattern : patterns) {
+ if (pattern.matcher(key).matches()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private byte[] encodeRecord(
+ final RecordSetWriterFactory writerFactory,
+ final MapRecord mapRecord) throws IOException {
+
+ try {
+ final RecordSchema writeSchema = writerFactory.getSchema(Collections.emptyMap(), PROVENANCE_SCHEMA);
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ try (final RecordSetWriter writer =
+ writerFactory.createWriter(getLogger(), writeSchema, baos, Collections.emptyMap())) {
+ writer.write(mapRecord);
+ }
+ return baos.toByteArray();
+ } catch (final SchemaNotFoundException e) {
+ throw new IOException("Schema lookup failed for Provenance record", e);
+ }
+ }
+
+ private ObjectNode serializeToJson(
+ final ProvenanceEventRecord event,
+ final String componentName,
+ final String processGroupId,
+ final String processGroupName,
+ final EncodingContext ctx) {
+
+ final boolean allowNullValues = ctx.allowNullValues();
+ final String contentBase = resolveContentBase(event, ctx);
+ final ObjectNode node = OBJECT_MAPPER.createObjectNode();
+
+ addField(node, FIELD_EVENT_ID, UUID.randomUUID().toString(), allowNullValues);
+ addField(node, FIELD_EVENT_ORDINAL, event.getEventId(), allowNullValues);
+ addField(node, FIELD_EVENT_TYPE, event.getEventType().name(), allowNullValues);
+ addField(node, FIELD_TIMESTAMP_MILLIS, event.getEventTime(), allowNullValues);
+ addField(node, FIELD_TIMESTAMP,
+ DATE_TIME_FORMATTER.format(Instant.ofEpochMilli(event.getEventTime())), allowNullValues);
+ addField(node, FIELD_DURATION_MILLIS, event.getEventDuration(), allowNullValues);
+ addField(node, FIELD_LINEAGE_START, event.getLineageStartDate(), allowNullValues);
+ addField(node, FIELD_DETAILS, event.getDetails(), allowNullValues);
+ addField(node, FIELD_COMPONENT_ID, event.getComponentId(), allowNullValues);
+ addField(node, FIELD_COMPONENT_TYPE, event.getComponentType(), allowNullValues);
+ addField(node, FIELD_COMPONENT_NAME, componentName, allowNullValues);
+ addField(node, FIELD_PROCESS_GROUP_ID, processGroupId, allowNullValues);
+ addField(node, FIELD_PROCESS_GROUP_NAME, processGroupName, allowNullValues);
+ addField(node, FIELD_ENTITY_ID, event.getFlowFileUuid(), allowNullValues);
+ addField(node, FIELD_ENTITY_TYPE, "org.apache.nifi.flowfile.FlowFile", allowNullValues);
+ addField(node, FIELD_ENTITY_SIZE, event.getFileSize(), allowNullValues);
+ addField(node, FIELD_PREV_ENTITY_SIZE, event.getPreviousFileSize(), allowNullValues);
+ addMapField(node, FIELD_UPDATED_ATTRIBUTES,
+ filterAttributes(event.getUpdatedAttributes()), allowNullValues);
+ addMapField(node, FIELD_PREV_ATTRIBUTES,
+ filterAttributes(event.getPreviousAttributes()), allowNullValues);
+ addField(node, FIELD_ACTOR_HOSTNAME, ctx.hostname(), allowNullValues);
+ if (contentBase != null) {
+ final String clusterSuffix = resolveClusterSuffix(ctx);
+ addField(node, FIELD_CONTENT_URI, contentBase + "output" + clusterSuffix, allowNullValues);
+ addField(node, FIELD_PREV_CONTENT_URI, contentBase + "input" + clusterSuffix, allowNullValues);
+ } else if (allowNullValues) {
+ node.putNull(FIELD_CONTENT_URI);
+ node.putNull(FIELD_PREV_CONTENT_URI);
+ }
+ addCollectionField(node, FIELD_PARENT_IDS, event.getParentUuids(), allowNullValues);
+ addCollectionField(node, FIELD_CHILD_IDS, event.getChildUuids(), allowNullValues);
+ addField(node, FIELD_TRANSIT_URI, event.getTransitUri(), allowNullValues);
+ addField(node, FIELD_REMOTE_IDENTIFIER, event.getSourceSystemFlowFileIdentifier(), allowNullValues);
+ addField(node, FIELD_ALT_IDENTIFIER, event.getAlternateIdentifierUri(), allowNullValues);
+ addField(node, FIELD_PLATFORM, ctx.platform(), allowNullValues);
+ addField(node, FIELD_APPLICATION, ctx.applicationName(), allowNullValues);
+
+ return node;
+ }
+
+ private ProducerConfiguration buildProducerConfiguration(final ConfigurationContext context) {
+ final boolean transactionsEnabled = context.getProperty(TRANSACTIONS_ENABLED).asBoolean();
+ final String transactionalIdPrefix = transactionsEnabled
+ ? context.getProperty(TRANSACTIONAL_ID_PREFIX).evaluateAttributeExpressions().getValue()
+ : null;
+ return new ProducerConfiguration(
+ transactionsEnabled,
+ transactionalIdPrefix,
+ context.getProperty(DELIVERY_GUARANTEE).getValue(),
+ context.getProperty(COMPRESSION_TYPE).getValue(),
+ null,
+ context.getProperty(MAX_REQUEST_SIZE).asDataSize(DataUnit.B).intValue()
+ );
+ }
+
+ private static byte[] resolveMessageKey(final ProvenanceEventRecord event, final String keyField) {
+ final String keyValue = switch (keyField) {
+ case FIELD_LINEAGE_START -> String.valueOf(event.getLineageStartDate());
+ case FIELD_ENTITY_ID -> event.getFlowFileUuid();
+ case FIELD_COMPONENT_ID -> event.getComponentId();
+ case FIELD_EVENT_TYPE -> event.getEventType().name();
+ case FIELD_EVENT_ORDINAL -> String.valueOf(event.getEventId());
+ default -> null;
+ };
+ return keyValue == null ? null : keyValue.getBytes(StandardCharsets.UTF_8);
+ }
+
+ private static String resolveUrlBase(final URL nifiUrl) {
+ final String urlString = nifiUrl.toString();
+ return urlString.endsWith(NIFI_API_PATH)
+ ? urlString.substring(0, urlString.length() - NIFI_API_PATH.length())
+ : urlString;
+ }
+
+ private static String resolveContentBase(final ProvenanceEventRecord event, final EncodingContext ctx) {
+ return ctx.nifiUrlBase() == null ? null
+ : ctx.nifiUrlBase() + "/nifi-api/provenance-events/" + event.getEventId() + "/content/";
+ }
+
+ private static String resolveClusterSuffix(final EncodingContext ctx) {
+ return ctx.nodeIdentifier() == null ? "" : "?clusterNodeId=" + ctx.nodeIdentifier();
+ }
+
+ private static void addField(final ObjectNode node, final String key,
+ final Object value, final boolean allowNullValues) {
+ switch (value) {
+ case String s -> node.put(key, s);
+ case Long l -> node.put(key, l);
+ case Integer i -> node.put(key, i);
+ case Boolean b -> node.put(key, b);
+ case null -> {
+ if (allowNullValues) {
+ node.putNull(key);
+ }
+ }
+ default -> node.put(key, value.toString());
+ }
+ }
+
+ private static void addMapField(final ObjectNode node, final String key,
+ final Map values,
+ final boolean allowNullValues) {
+ if (values != null) {
+ final ObjectNode mapNode = OBJECT_MAPPER.createObjectNode();
+ for (final Map.Entry entry : values.entrySet()) {
+ if (entry.getKey() == null) {
+ continue;
+ }
+ if (entry.getValue() == null) {
+ if (allowNullValues) {
+ mapNode.putNull(entry.getKey());
+ }
+ } else {
+ mapNode.put(entry.getKey(), entry.getValue());
+ }
+ }
+ node.set(key, mapNode);
+ } else if (allowNullValues) {
+ node.putNull(key);
+ }
+ }
+
+ private static void addCollectionField(final ObjectNode node, final String key,
+ final Collection values,
+ final boolean allowNullValues) {
+ if (values != null) {
+ final ArrayNode arrayNode = OBJECT_MAPPER.createArrayNode();
+ for (final String v : values) {
+ if (v != null) {
+ arrayNode.add(v);
+ }
+ }
+ node.set(key, arrayNode);
+ } else if (allowNullValues) {
+ node.putNull(key);
+ }
+ }
+}
diff --git a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-provenance-reporting-task/src/main/resources/META-INF/services/org.apache.nifi.reporting.ReportingTask b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-provenance-reporting-task/src/main/resources/META-INF/services/org.apache.nifi.reporting.ReportingTask
new file mode 100644
index 000000000000..a74433e6f673
--- /dev/null
+++ b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-provenance-reporting-task/src/main/resources/META-INF/services/org.apache.nifi.reporting.ReportingTask
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.nifi.kafka.reporting.KafkaProvenanceReportingTask
diff --git a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-provenance-reporting-task/src/main/resources/docs/org.apache.nifi.kafka.reporting.KafkaProvenanceReportingTask/additionalDetails.md b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-provenance-reporting-task/src/main/resources/docs/org.apache.nifi.kafka.reporting.KafkaProvenanceReportingTask/additionalDetails.md
new file mode 100644
index 000000000000..c5376fd164fa
--- /dev/null
+++ b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-provenance-reporting-task/src/main/resources/docs/org.apache.nifi.kafka.reporting.KafkaProvenanceReportingTask/additionalDetails.md
@@ -0,0 +1,217 @@
+
+
+# KafkaProvenanceReportingTask
+
+The Kafka Provenance Reporting Task publishes NiFi Provenance Events directly to a Kafka topic using the Kafka
+Connection Service. On each scheduling trigger the task reads a configurable batch of events from the provenance
+repository and publishes them to Kafka. The ID of the last published event is stored in local state so that the
+task resumes from the correct position after a restart.
+
+Events can be filtered by event type, component type, component name, and component ID before publishing.
+
+A configurable message key field controls how Kafka assigns events to partitions, which is useful for grouping
+related events (e.g. all events for the same FlowFile or lineage) on the same partition for ordered consumption.
+
+Each event is published as an individual Kafka message. By default events are serialized as JSON objects. When a
+Record Writer controller service is configured, the serialization format is delegated to that writer, allowing
+formats such as Avro or Parquet. The user can also control which fields are written by defining a schema on the
+Record Writer (e.g. a subset of the full reporting task schema), giving full control over the output format and
+data. The record schema used as input to the Record Writer is defined as follows:
+
+```json
+{
+ "type": "record",
+ "name": "provenance",
+ "namespace": "provenance",
+ "fields": [
+ {
+ "name": "eventId",
+ "type": "string"
+ },
+ {
+ "name": "eventOrdinal",
+ "type": "long"
+ },
+ {
+ "name": "eventType",
+ "type": "string"
+ },
+ {
+ "name": "timestampMillis",
+ "type": "long"
+ },
+ {
+ "name": "durationMillis",
+ "type": "long"
+ },
+ {
+ "name": "lineageStart",
+ "type": {
+ "type": "long",
+ "logicalType": "timestamp-millis"
+ }
+ },
+ {
+ "name": "details",
+ "type": [
+ "null",
+ "string"
+ ]
+ },
+ {
+ "name": "componentId",
+ "type": [
+ "null",
+ "string"
+ ]
+ },
+ {
+ "name": "componentType",
+ "type": [
+ "null",
+ "string"
+ ]
+ },
+ {
+ "name": "componentName",
+ "type": [
+ "null",
+ "string"
+ ]
+ },
+ {
+ "name": "processGroupId",
+ "type": [
+ "null",
+ "string"
+ ]
+ },
+ {
+ "name": "processGroupName",
+ "type": [
+ "null",
+ "string"
+ ]
+ },
+ {
+ "name": "entityId",
+ "type": [
+ "null",
+ "string"
+ ]
+ },
+ {
+ "name": "entityType",
+ "type": [
+ "null",
+ "string"
+ ]
+ },
+ {
+ "name": "entitySize",
+ "type": [
+ "null",
+ "long"
+ ]
+ },
+ {
+ "name": "previousEntitySize",
+ "type": [
+ "null",
+ "long"
+ ]
+ },
+ {
+ "name": "updatedAttributes",
+ "type": {
+ "type": "map",
+ "values": "string"
+ }
+ },
+ {
+ "name": "previousAttributes",
+ "type": {
+ "type": "map",
+ "values": "string"
+ }
+ },
+ {
+ "name": "actorHostname",
+ "type": [
+ "null",
+ "string"
+ ]
+ },
+ {
+ "name": "contentURI",
+ "type": [
+ "null",
+ "string"
+ ]
+ },
+ {
+ "name": "previousContentURI",
+ "type": [
+ "null",
+ "string"
+ ]
+ },
+ {
+ "name": "parentIds",
+ "type": {
+ "type": "array",
+ "items": "string"
+ }
+ },
+ {
+ "name": "childIds",
+ "type": {
+ "type": "array",
+ "items": "string"
+ }
+ },
+ {
+ "name": "platform",
+ "type": "string"
+ },
+ {
+ "name": "application",
+ "type": "string"
+ },
+ {
+ "name": "remoteIdentifier",
+ "type": [
+ "null",
+ "string"
+ ]
+ },
+ {
+ "name": "alternateIdentifier",
+ "type": [
+ "null",
+ "string"
+ ]
+ },
+ {
+ "name": "transitUri",
+ "type": [
+ "null",
+ "string"
+ ]
+ }
+ ]
+}
+```
\ No newline at end of file
diff --git a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-provenance-reporting-task/src/test/java/org/apache/nifi/kafka/reporting/TestKafkaProvenanceReportingTask.java b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-provenance-reporting-task/src/test/java/org/apache/nifi/kafka/reporting/TestKafkaProvenanceReportingTask.java
new file mode 100644
index 000000000000..feb9afdb0a45
--- /dev/null
+++ b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-provenance-reporting-task/src/test/java/org/apache/nifi/kafka/reporting/TestKafkaProvenanceReportingTask.java
@@ -0,0 +1,950 @@
+/*
+ * 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
+ * https://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.reporting;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.PropertyValue;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.controller.status.ProcessGroupStatus;
+import org.apache.nifi.controller.status.ProcessorStatus;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.kafka.service.api.KafkaConnectionService;
+import org.apache.nifi.kafka.service.api.producer.KafkaProducerService;
+import org.apache.nifi.kafka.service.api.producer.ProducerConfiguration;
+import org.apache.nifi.kafka.service.api.record.KafkaRecord;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.provenance.ProvenanceEventBuilder;
+import org.apache.nifi.provenance.ProvenanceEventRecord;
+import org.apache.nifi.provenance.ProvenanceEventRepository;
+import org.apache.nifi.provenance.ProvenanceEventType;
+import org.apache.nifi.provenance.StandardProvenanceEventRecord;
+import org.apache.nifi.reporting.EventAccess;
+import org.apache.nifi.reporting.ReportingContext;
+import org.apache.nifi.reporting.ReportingInitializationContext;
+import org.apache.nifi.state.MockStateManager;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.MockPropertyValue;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+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.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class TestKafkaProvenanceReportingTask {
+
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ private ReportingContext context;
+ private ReportingInitializationContext initContext;
+ private ConfigurationContext confContext;
+ private KafkaProducerService producerService;
+
+ private List sentRecords;
+
+ @BeforeEach
+ void setUp() {
+ context = mock(ReportingContext.class);
+ initContext = mock(ReportingInitializationContext.class);
+ confContext = mock(ConfigurationContext.class);
+ producerService = mock(KafkaProducerService.class);
+ sentRecords = new ArrayList<>();
+
+ when(producerService.isClosed()).thenReturn(false);
+ doAnswer(inv -> {
+ @SuppressWarnings("unchecked") final Iterator it = (Iterator) inv.getArgument(0);
+ it.forEachRemaining(sentRecords::add);
+ return null;
+ }).when(producerService).send(any(), any());
+ }
+
+ private Map defaultProperties() {
+ final KafkaProvenanceReportingTask task = new KafkaProvenanceReportingTask();
+ final Map properties = new HashMap<>();
+ for (final PropertyDescriptor pd : task.getSupportedPropertyDescriptors()) {
+ properties.put(pd, pd.getDefaultValue());
+ }
+ properties.put(KafkaProvenanceReportingTask.INSTANCE_URL, "https://localhost:8443/nifi");
+ properties.put(KafkaProvenanceReportingTask.PLATFORM, "nifi");
+ properties.put(KafkaProvenanceReportingTask.TOPIC_NAME, "provenance-events");
+ return properties;
+ }
+
+ private KafkaProvenanceReportingTask setup(
+ final ProvenanceEventRecord event,
+ final Map properties,
+ final long maxEventId) throws IOException {
+
+ final KafkaProvenanceReportingTask task = new KafkaProvenanceReportingTask();
+
+ // State manager
+ when(context.getStateManager()).thenReturn(new MockStateManager(task));
+
+ // ReportingContext properties (used in onTrigger)
+ doAnswer(inv -> {
+ final PropertyDescriptor pd = inv.getArgument(0, PropertyDescriptor.class);
+ return new MockPropertyValue(properties.get(pd));
+ }).when(context).getProperty(any(PropertyDescriptor.class));
+
+ // ConfigurationContext properties (used in onScheduled)
+ // KAFKA_CONNECTION_SERVICE requires a controller-service mock; everything else uses MockPropertyValue.
+ final KafkaConnectionService connectionService = mock(KafkaConnectionService.class);
+ when(connectionService.getProducerService(any(ProducerConfiguration.class))).thenReturn(producerService);
+
+ doAnswer(inv -> {
+ final PropertyDescriptor pd = inv.getArgument(0, PropertyDescriptor.class);
+ if (KafkaProvenanceReportingTask.KAFKA_CONNECTION_SERVICE.equals(pd)) {
+ final PropertyValue pv = mock(PropertyValue.class);
+ doReturn(connectionService).when(pv).asControllerService(KafkaConnectionService.class);
+ return pv;
+ }
+ return new MockPropertyValue(properties.get(pd));
+ }).when(confContext).getProperty(any(PropertyDescriptor.class));
+
+ // EventAccess: returns the test event until maxEventId total events have been delivered
+ final AtomicInteger totalEvents = new AtomicInteger(0);
+ final EventAccess eventAccess = mock(EventAccess.class);
+ doAnswer(inv -> {
+ final long startId = inv.getArgument(0, Long.class);
+ final int maxRecords = inv.getArgument(1, Integer.class);
+ final List result = new ArrayList<>();
+ for (int i = (int) Math.max(0, startId);
+ i < startId + maxRecords && totalEvents.get() < maxEventId;
+ i++) {
+ if (event != null) {
+ result.add(event);
+ }
+ totalEvents.getAndIncrement();
+ }
+ return result;
+ }).when(eventAccess).getProvenanceEvents(anyLong(), anyInt());
+
+ // Process group hierarchy: root => processor "processor-1" / "Test Processor"
+ final ProcessGroupStatus pgRoot = new ProcessGroupStatus();
+ pgRoot.setId("root");
+ pgRoot.setName("NiFi Flow");
+
+ final ProcessorStatus prcRoot = new ProcessorStatus();
+ prcRoot.setId("processor-1");
+ prcRoot.setName("Test Processor");
+ pgRoot.getProcessorStatus().add(prcRoot);
+
+ when(eventAccess.getControllerStatus()).thenReturn(pgRoot);
+
+ final ProvenanceEventRepository provenanceRepository = mock(ProvenanceEventRepository.class);
+ doAnswer(inv -> maxEventId).when(provenanceRepository).getMaxEventId();
+ when(eventAccess.getProvenanceRepository()).thenReturn(provenanceRepository);
+ when(context.getEventAccess()).thenReturn(eventAccess);
+
+ // Cluster (standalone by default; override per test for cluster scenarios)
+ when(context.isClustered()).thenReturn(false);
+ when(context.getClusterNodeIdentifier()).thenReturn(null);
+
+ // Logger
+ final ComponentLog logger = mock(ComponentLog.class);
+ when(initContext.getIdentifier()).thenReturn("test-task-id");
+ when(initContext.getLogger()).thenReturn(logger);
+
+ return task;
+ }
+
+ private ProvenanceEventRecord createEvent() {
+ return createEvent("processor-1", "TestProcessor");
+ }
+
+ private ProvenanceEventRecord createEvent(final String componentId, final String componentType) {
+ final String uuid = UUID.randomUUID().toString();
+ final Map attributes = new HashMap<>();
+ attributes.put("uuid", uuid);
+ attributes.put("filename", "test.txt");
+ attributes.put("nullAttr", null);
+
+ final ProvenanceEventBuilder builder = new StandardProvenanceEventRecord.Builder();
+ builder.setEventTime(System.currentTimeMillis());
+ builder.setEventType(ProvenanceEventType.RECEIVE);
+ builder.setTransitUri("nifi://unit-test");
+ builder.fromFlowFile(createFlowFile(1L, attributes));
+ builder.setAttributes(Collections.emptyMap(), attributes);
+ builder.setComponentId(componentId);
+ builder.setComponentType(componentType);
+ return builder.build();
+ }
+
+ private FlowFile createFlowFile(final long id, final Map attributes) {
+ final MockFlowFile flowFile = new MockFlowFile(id);
+ flowFile.putAttributes(attributes);
+ return flowFile;
+ }
+
+ private ObjectNode parseJson(final byte[] payload) throws IOException {
+ return (ObjectNode) OBJECT_MAPPER.readTree(payload);
+ }
+
+ @Test
+ void testJsonSerializationCoreFields() throws Exception {
+ final Map props = defaultProperties();
+ final ProvenanceEventRecord event = createEvent();
+
+ final KafkaProvenanceReportingTask task = setup(event, props, 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(1, sentRecords.size(), "Expected exactly one Kafka record");
+ final ObjectNode json = parseJson(sentRecords.getFirst().getValue());
+
+ // Identity fields
+ assertNotNull(json.get("eventId"), "eventId must be present");
+ assertEquals("RECEIVE", json.get("eventType").asText());
+ assertNotNull(json.get("eventOrdinal"), "eventOrdinal must be present");
+ assertNotNull(json.get("timestampMillis"), "timestampMillis must be present");
+ assertNotNull(json.get("timestamp"), "timestamp (ISO-8601) must be present");
+ assertNotNull(json.get("durationMillis"), "durationMillis must be present");
+
+ // Component fields
+ assertEquals("processor-1", json.get("componentId").asText());
+ assertEquals("TestProcessor", json.get("componentType").asText());
+ assertEquals("Test Processor", json.get("componentName").asText(),
+ "componentName resolved from process group status");
+ assertEquals("org.apache.nifi.flowfile.FlowFile", json.get("entityType").asText());
+
+ // Platform / application
+ assertEquals("nifi", json.get("platform").asText());
+ assertEquals("NiFi Flow", json.get("application").asText(),
+ "application is the root process group name");
+ }
+
+ @Test
+ void testJsonTimestampIsIso8601() throws Exception {
+ final KafkaProvenanceReportingTask task = setup(createEvent(), defaultProperties(), 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(1, sentRecords.size());
+ final String timestamp = parseJson(sentRecords.getFirst().getValue()).get("timestamp").asText();
+ // Must parse as a valid instant (e.g. "2025-04-28T10:30:00.000Z")
+ assertDoesNotThrow(() -> Instant.parse(timestamp),
+ "timestamp must be a parseable ISO-8601 instant: " + timestamp);
+ assertTrue(timestamp.endsWith("Z"), "timestamp must be UTC (ends with Z)");
+ }
+
+ @Test
+ void testJsonNullValuesOmittedByDefault() throws Exception {
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.ALLOW_NULL_VALUES, "false");
+
+ // Event has no 'details' set; it will be null in the record
+ final KafkaProvenanceReportingTask task = setup(createEvent(), props, 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(1, sentRecords.size());
+ final ObjectNode json = parseJson(sentRecords.getFirst().getValue());
+
+ assertFalse(json.has("details"),
+ "Null 'details' field must be omitted when allowNullValues=false");
+ assertFalse(json.has("remoteIdentifier"),
+ "Null 'remoteIdentifier' field must be omitted when allowNullValues=false");
+ assertFalse(json.has("alternateIdentifier"),
+ "Null 'alternateIdentifier' field must be omitted when allowNullValues=false");
+ }
+
+ @Test
+ void testJsonNullValuesIncludedWhenEnabled() throws Exception {
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.ALLOW_NULL_VALUES, "true");
+
+ final KafkaProvenanceReportingTask task = setup(createEvent(), props, 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(1, sentRecords.size());
+ final ObjectNode json = parseJson(sentRecords.getFirst().getValue());
+
+ assertTrue(json.has("details") && json.get("details").isNull(),
+ "Null 'details' field must appear as JSON null when allowNullValues=true");
+ assertTrue(json.has("remoteIdentifier") && json.get("remoteIdentifier").isNull(),
+ "Null 'remoteIdentifier' must appear as JSON null when allowNullValues=true");
+ }
+
+ @Test
+ void testJsonUpdatedAttributesSerialized() throws Exception {
+ final KafkaProvenanceReportingTask task = setup(createEvent(), defaultProperties(), 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(1, sentRecords.size());
+ final ObjectNode json = parseJson(sentRecords.getFirst().getValue());
+ final ObjectNode attrs = (ObjectNode) json.get("updatedAttributes");
+
+ assertNotNull(attrs, "updatedAttributes must be present");
+ assertEquals("test.txt", attrs.get("filename").asText());
+ // null-valued attribute "nullAttr" omitted because allowNullValues defaults to false
+ assertFalse(attrs.has("nullAttr"));
+ }
+
+ @Test
+ void testMessageKeyLineageStart() throws Exception {
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.MESSAGE_KEY_FIELD,
+ KafkaProvenanceReportingTask.KEY_LINEAGE_START.getValue());
+
+ final KafkaProvenanceReportingTask task = setup(createEvent(), props, 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(1, sentRecords.size());
+ final byte[] key = sentRecords.getFirst().getKey();
+ assertNotNull(key, "Message key must be set for lineageStart");
+ // Key must be parseable as a long (epoch millis)
+ assertDoesNotThrow(() -> Long.parseLong(new String(key, StandardCharsets.UTF_8)),
+ "lineageStart key must be a numeric epoch-millis string");
+ }
+
+ @Test
+ void testMessageKeyEntityId() throws Exception {
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.MESSAGE_KEY_FIELD,
+ KafkaProvenanceReportingTask.KEY_ENTITY_ID.getValue());
+
+ final ProvenanceEventRecord event = createEvent();
+ final KafkaProvenanceReportingTask task = setup(event, props, 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(1, sentRecords.size());
+ final String key = new String(sentRecords.getFirst().getKey(), StandardCharsets.UTF_8);
+ assertEquals(event.getFlowFileUuid(), key, "Key must equal the FlowFile UUID");
+ }
+
+ @Test
+ void testMessageKeyNone() throws Exception {
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.MESSAGE_KEY_FIELD,
+ KafkaProvenanceReportingTask.KEY_NONE.getValue());
+
+ final KafkaProvenanceReportingTask task = setup(createEvent(), props, 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(1, sentRecords.size());
+ assertNull(sentRecords.getFirst().getKey(), "No key must be set when KEY_NONE is selected");
+ }
+
+ @Test
+ void testFilterIncludeEventTypeAllowsMatchingEvents() throws Exception {
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.FILTER_EVENT_TYPE, "RECEIVE");
+
+ final KafkaProvenanceReportingTask task = setup(createEvent(), props, 3);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(3, sentRecords.size(),
+ "All RECEIVE events must pass when RECEIVE is the include filter");
+ }
+
+ @Test
+ void testFilterIncludeEventTypeFiltersOutNonMatchingEvents() throws Exception {
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.FILTER_EVENT_TYPE, "DROP");
+
+ final KafkaProvenanceReportingTask task = setup(createEvent(), props, 3);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(0, sentRecords.size(),
+ "RECEIVE events must be filtered out when only DROP is included");
+ }
+
+ @Test
+ void testFilterExcludeEventTypeFiltersOutMatchingEvents() throws Exception {
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.FILTER_EVENT_TYPE_EXCLUDE, "RECEIVE");
+
+ final KafkaProvenanceReportingTask task = setup(createEvent(), props, 3);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(0, sentRecords.size(),
+ "RECEIVE events must be excluded when RECEIVE is in the exclude filter");
+ }
+
+ @Test
+ void testFilterExcludeTakesPrecedenceOverInclude() throws Exception {
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.FILTER_EVENT_TYPE, "RECEIVE");
+ props.put(KafkaProvenanceReportingTask.FILTER_EVENT_TYPE_EXCLUDE, "RECEIVE");
+
+ final KafkaProvenanceReportingTask task = setup(createEvent(), props, 3);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(0, sentRecords.size(),
+ "Exclude filter must take precedence over include filter");
+ }
+
+ @Test
+ void testFilterIncludeComponentTypeAllowsMatchingEvents() throws Exception {
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.FILTER_COMPONENT_TYPE, "Test.*");
+
+ final KafkaProvenanceReportingTask task = setup(createEvent("processor-1", "TestProcessor"), props, 2);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(2, sentRecords.size(),
+ "Events from 'TestProcessor' must pass the 'Test.*' include regex");
+ }
+
+ @Test
+ void testFilterIncludeComponentTypeFiltersOutNonMatchingEvents() throws Exception {
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.FILTER_COMPONENT_TYPE, "SomeOther.*");
+
+ final KafkaProvenanceReportingTask task = setup(createEvent("processor-1", "TestProcessor"), props, 2);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(0, sentRecords.size(),
+ "Events from 'TestProcessor' must be filtered out by the 'SomeOther.*' include regex");
+ }
+
+ @Test
+ void testFilterExcludeComponentTypeFiltersOutMatchingEvents() throws Exception {
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.FILTER_COMPONENT_TYPE_EXCLUDE, "Test.*");
+
+ final KafkaProvenanceReportingTask task = setup(createEvent("processor-1", "TestProcessor"), props, 2);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(0, sentRecords.size(),
+ "Events from 'TestProcessor' must be excluded by the 'Test.*' exclude regex");
+ }
+
+ @Test
+ void testFilterIncludeComponentIdAllowsMatchingEvents() throws Exception {
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.FILTER_COMPONENT_ID, "processor-1");
+
+ final KafkaProvenanceReportingTask task = setup(createEvent("processor-1", "TestProcessor"), props, 2);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(2, sentRecords.size(),
+ "Events from 'processor-1' must pass when it is the include ID");
+ }
+
+ @Test
+ void testFilterIncludeComponentIdFiltersOutNonMatchingEvents() throws Exception {
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.FILTER_COMPONENT_ID, "other-processor");
+
+ final KafkaProvenanceReportingTask task = setup(createEvent("processor-1", "TestProcessor"), props, 2);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(0, sentRecords.size(),
+ "Events from 'processor-1' must be filtered out when 'other-processor' is included");
+ }
+
+ @Test
+ void testFilterExcludeComponentIdFiltersOutMatchingEvents() throws Exception {
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.FILTER_COMPONENT_ID_EXCLUDE, "processor-1");
+
+ final KafkaProvenanceReportingTask task = setup(createEvent("processor-1", "TestProcessor"), props, 2);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(0, sentRecords.size(),
+ "Events from 'processor-1' must be excluded when it is in the exclude ID list");
+ }
+
+ @Test
+ void testSkipTriggerWhenClusteredWithoutNodeId() throws Exception {
+ final KafkaProvenanceReportingTask task = setup(createEvent(), defaultProperties(), 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+
+ // Simulate clustered node that hasn't received its node ID yet
+ when(context.isClustered()).thenReturn(true);
+ when(context.getClusterNodeIdentifier()).thenReturn(null);
+ task.onTrigger(context);
+
+ assertEquals(0, sentRecords.size(),
+ "Trigger must be skipped when running clustered but node ID is not yet available");
+ }
+
+ @Test
+ void testSkipTriggerWhenProducerClosed() throws Exception {
+ // Signal the producer as closed before it is used
+ when(producerService.isClosed()).thenReturn(true);
+
+ final KafkaProvenanceReportingTask task = setup(createEvent(), defaultProperties(), 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(0, sentRecords.size(),
+ "Trigger must be skipped when the KafkaProducerService is closed");
+ }
+
+ @Test
+ void testContentUriPresentWhenInstanceUrlConfigured() throws Exception {
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.INSTANCE_URL, "https://localhost:8443/nifi");
+
+ final KafkaProvenanceReportingTask task = setup(createEvent(), props, 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(1, sentRecords.size());
+ final ObjectNode json = parseJson(sentRecords.getFirst().getValue());
+
+ assertTrue(json.has("contentURI"), "contentURI must be present");
+ assertTrue(json.has("previousContentURI"), "previousContentURI must be present");
+ assertTrue(json.get("contentURI").asText().contains("/nifi-api/provenance-events/"),
+ "contentURI must contain the NiFi API path");
+ assertTrue(json.get("contentURI").asText().endsWith("/content/output"),
+ "contentURI must end with /content/output");
+ assertTrue(json.get("previousContentURI").asText().endsWith("/content/input"),
+ "previousContentURI must end with /content/input");
+ }
+
+ @Test
+ void testContentUriAbsentWhenInstanceUrlBlank() throws Exception {
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.INSTANCE_URL, "");
+ props.put(KafkaProvenanceReportingTask.ALLOW_NULL_VALUES, "false");
+
+ final KafkaProvenanceReportingTask task = setup(createEvent(), props, 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(1, sentRecords.size());
+ final ObjectNode json = parseJson(sentRecords.getFirst().getValue());
+
+ assertFalse(json.has("contentURI"),
+ "contentURI must be absent when no instance URL is configured");
+ assertFalse(json.has("previousContentURI"),
+ "previousContentURI must be absent when no instance URL is configured");
+ }
+
+ @Test
+ void testCustomValidateBothWhitelistAndBlacklistFails() {
+ final KafkaProvenanceReportingTask task = new KafkaProvenanceReportingTask();
+ final ValidationContext validationContext = mock(ValidationContext.class);
+ doAnswer(inv -> {
+ final PropertyDescriptor pd = inv.getArgument(0, PropertyDescriptor.class);
+ if (KafkaProvenanceReportingTask.FILTER_ATTRIBUTES_INCLUDE.equals(pd)) {
+ return new MockPropertyValue("foo");
+ }
+ if (KafkaProvenanceReportingTask.FILTER_ATTRIBUTES_EXCLUDE.equals(pd)) {
+ return new MockPropertyValue("bar");
+ }
+ return new MockPropertyValue(null);
+ }).when(validationContext).getProperty(any(PropertyDescriptor.class));
+
+ final Collection results = task.customValidate(validationContext);
+ assertFalse(results.isEmpty(),
+ "customValidate must return at least one result when both whitelist and blacklist are set");
+ assertTrue(results.stream().anyMatch(r -> !r.isValid()),
+ "At least one validation result must indicate failure when both filters are configured");
+ }
+
+ @Test
+ void testCustomValidateOnlyWhitelistIsValid() {
+ final KafkaProvenanceReportingTask task = new KafkaProvenanceReportingTask();
+ final ValidationContext validationContext = mock(ValidationContext.class);
+ doAnswer(inv -> {
+ final PropertyDescriptor pd = inv.getArgument(0, PropertyDescriptor.class);
+ if (KafkaProvenanceReportingTask.FILTER_ATTRIBUTES_INCLUDE.equals(pd)) {
+ return new MockPropertyValue("foo");
+ }
+ return new MockPropertyValue(null);
+ }).when(validationContext).getProperty(any(PropertyDescriptor.class));
+
+ final Collection results = task.customValidate(validationContext);
+ assertTrue(results.isEmpty() || results.stream().allMatch(ValidationResult::isValid),
+ "customValidate must pass when only whitelist is configured");
+ }
+
+ @Test
+ void testCustomValidateOnlyBlacklistIsValid() {
+ final KafkaProvenanceReportingTask task = new KafkaProvenanceReportingTask();
+ final ValidationContext validationContext = mock(ValidationContext.class);
+ doAnswer(inv -> {
+ final PropertyDescriptor pd = inv.getArgument(0, PropertyDescriptor.class);
+ if (KafkaProvenanceReportingTask.FILTER_ATTRIBUTES_EXCLUDE.equals(pd)) {
+ return new MockPropertyValue("bar");
+ }
+ return new MockPropertyValue(null);
+ }).when(validationContext).getProperty(any(PropertyDescriptor.class));
+
+ final Collection results = task.customValidate(validationContext);
+ assertTrue(results.isEmpty() || results.stream().allMatch(ValidationResult::isValid),
+ "customValidate must pass when only blacklist is configured");
+ }
+
+ private ProvenanceEventRecord createEventWithAttributes(
+ final Map previousAttributes,
+ final Map updatedAttributes) {
+ final ProvenanceEventBuilder builder = new StandardProvenanceEventRecord.Builder();
+ builder.setEventTime(System.currentTimeMillis());
+ builder.setEventType(ProvenanceEventType.RECEIVE);
+ builder.setTransitUri("nifi://unit-test");
+ builder.fromFlowFile(createFlowFile(1L, updatedAttributes));
+ builder.setAttributes(previousAttributes, updatedAttributes);
+ builder.setComponentId("processor-1");
+ builder.setComponentType("TestProcessor");
+ return builder.build();
+ }
+
+ @Test
+ void testAttributeWhitelistLiteralNamesIncludesOnlyMatchingAttributes() throws Exception {
+ final Map attrs = new HashMap<>();
+ attrs.put("foo", "foo-value");
+ attrs.put("bar", "bar-value");
+ attrs.put("baz", "baz-value");
+ final ProvenanceEventRecord event = createEventWithAttributes(attrs, attrs);
+
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.FILTER_ATTRIBUTES_INCLUDE, "foo,bar");
+
+ final KafkaProvenanceReportingTask task = setup(event, props, 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertFalse(task.attributeIncludePatterns.get().isEmpty(), "Whitelist patterns must be compiled in onScheduled");
+ assertEquals(2, task.attributeIncludePatterns.get().size(), "Must have exactly 2 compiled patterns");
+
+ assertEquals(1, sentRecords.size());
+ final ObjectNode json = (ObjectNode) parseJson(sentRecords.getFirst().getValue()).get("updatedAttributes");
+ assertTrue(json.has("foo"), "Whitelisted attribute 'foo' must be present");
+ assertTrue(json.has("bar"), "Whitelisted attribute 'bar' must be present");
+ assertFalse(json.has("baz"), "Non-whitelisted attribute 'baz' must be absent");
+ }
+
+ @Test
+ void testAttributeWhitelistRegexPatternIncludesMatchingAttributes() throws Exception {
+ final Map attrs = new HashMap<>();
+ attrs.put("user.id", "u-001");
+ attrs.put("user.name", "alice");
+ attrs.put("filename", "data.csv");
+ final ProvenanceEventRecord event = createEventWithAttributes(attrs, attrs);
+
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.FILTER_ATTRIBUTES_INCLUDE, "user\\..*");
+
+ final KafkaProvenanceReportingTask task = setup(event, props, 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(1, sentRecords.size());
+ final ObjectNode json = (ObjectNode) parseJson(sentRecords.getFirst().getValue()).get("updatedAttributes");
+ assertTrue(json.has("user.id"), "Attribute matching regex must be included");
+ assertTrue(json.has("user.name"), "Attribute matching regex must be included");
+ assertFalse(json.has("filename"), "Non-matching attribute must be excluded");
+ }
+
+ @Test
+ void testAttributeBlacklistLiteralNamesExcludesMatchingAttributes() throws Exception {
+ final Map attrs = new HashMap<>();
+ attrs.put("foo", "foo-value");
+ attrs.put("bar", "bar-value");
+ attrs.put("baz", "baz-value");
+ final ProvenanceEventRecord event = createEventWithAttributes(attrs, attrs);
+
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.FILTER_ATTRIBUTES_EXCLUDE, "foo,bar");
+
+ final KafkaProvenanceReportingTask task = setup(event, props, 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertFalse(task.attributeExcludePatterns.get().isEmpty(), "Blacklist patterns must be compiled in onScheduled");
+ assertEquals(2, task.attributeExcludePatterns.get().size(), "Must have exactly 2 compiled patterns");
+
+ assertEquals(1, sentRecords.size());
+ final ObjectNode json = (ObjectNode) parseJson(sentRecords.getFirst().getValue()).get("updatedAttributes");
+ assertFalse(json.has("foo"), "Blacklisted attribute 'foo' must be excluded");
+ assertFalse(json.has("bar"), "Blacklisted attribute 'bar' must be excluded");
+ assertTrue(json.has("baz"), "Non-blacklisted attribute 'baz' must be included");
+ }
+
+ @Test
+ void testAttributeBlacklistRegexPatternExcludesMatchingAttributes() throws Exception {
+ final Map attrs = new HashMap<>();
+ attrs.put("internal.token", "secret");
+ attrs.put("internal.key", "private");
+ attrs.put("filename", "data.csv");
+ final ProvenanceEventRecord event = createEventWithAttributes(attrs, attrs);
+
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.FILTER_ATTRIBUTES_EXCLUDE, "internal\\..*");
+
+ final KafkaProvenanceReportingTask task = setup(event, props, 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(1, sentRecords.size());
+ final ObjectNode json = (ObjectNode) parseJson(sentRecords.getFirst().getValue()).get("updatedAttributes");
+ assertFalse(json.has("internal.token"), "Blacklisted attribute must be excluded");
+ assertFalse(json.has("internal.key"), "Blacklisted attribute must be excluded");
+ assertTrue(json.has("filename"), "Non-blacklisted attribute must be included");
+ }
+
+ @Test
+ void testNoAttributeFilterPreservesAllAttributes() throws Exception {
+ final Map attrs = new HashMap<>();
+ attrs.put("foo", "foo-value");
+ attrs.put("bar", "bar-value");
+ final ProvenanceEventRecord event = createEventWithAttributes(attrs, attrs);
+
+ final KafkaProvenanceReportingTask task = setup(event, defaultProperties(), 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertTrue(task.attributeIncludePatterns.get().isEmpty(), "Whitelist patterns must be empty when whitelist is not configured");
+ assertTrue(task.attributeExcludePatterns.get().isEmpty(), "Blacklist patterns must be empty when blacklist is not configured");
+
+ assertEquals(1, sentRecords.size());
+ final ObjectNode json = (ObjectNode) parseJson(sentRecords.getFirst().getValue()).get("updatedAttributes");
+ assertTrue(json.has("foo"), "All attributes must be present when no filter is configured");
+ assertTrue(json.has("bar"), "All attributes must be present when no filter is configured");
+ }
+
+ @Test
+ void testAttributeWhitelistFilterAppliesToBothPreviousAndUpdatedAttributes() throws Exception {
+ final Map prevAttrs = new HashMap<>();
+ prevAttrs.put("keep", "prev-keep-val");
+ prevAttrs.put("drop", "prev-drop-val");
+
+ final Map updatedAttrs = new HashMap<>();
+ updatedAttrs.put("keep", "updated-keep-val");
+ updatedAttrs.put("drop", "updated-drop-val");
+
+ final ProvenanceEventRecord event = createEventWithAttributes(prevAttrs, updatedAttrs);
+
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.FILTER_ATTRIBUTES_INCLUDE, "keep");
+
+ final KafkaProvenanceReportingTask task = setup(event, props, 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(1, sentRecords.size());
+ final ObjectNode json = parseJson(sentRecords.getFirst().getValue());
+
+ final ObjectNode updated = (ObjectNode) json.get("updatedAttributes");
+ assertTrue(updated.has("keep"), "Whitelisted key must be present in updatedAttributes");
+ assertFalse(updated.has("drop"), "Non-whitelisted key must be absent from updatedAttributes");
+
+ final ObjectNode previous = (ObjectNode) json.get("previousAttributes");
+ assertTrue(previous.has("keep"), "Whitelisted key must be present in previousAttributes");
+ assertFalse(previous.has("drop"), "Non-whitelisted key must be absent from previousAttributes");
+ }
+
+ @Test
+ void testInvalidRegexInAttributeFilterFailsPropertyValidation() {
+ final Validator validator = KafkaProvenanceReportingTask.FILTER_ATTRIBUTES_INCLUDE.getValidators().getFirst();
+
+ final ValidationResult valid = validator.validate("Attribute Whitelist", "foo,bar.*", null);
+ assertTrue(valid.isValid(), "Valid patterns must pass the property validator");
+
+ final ValidationResult invalid = validator.validate("Attribute Whitelist", "foo,[invalid", null);
+ assertFalse(invalid.isValid(), "A malformed regex entry must fail the property validator");
+ assertTrue(invalid.getExplanation().contains("[invalid"),
+ "Validation explanation must name the offending entry");
+ }
+
+ @Test
+ void testAttributeMaxLengthTrimsLongValues() throws Exception {
+ final Map attrs = new HashMap<>();
+ attrs.put("short", "abc");
+ attrs.put("long", "a".repeat(200));
+ final ProvenanceEventRecord event = createEventWithAttributes(attrs, attrs);
+
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.ATTRIBUTE_MAX_LENGTH, "50");
+
+ final KafkaProvenanceReportingTask task = setup(event, props, 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(50, task.attributeMaxLength, "Effective max length must match configured value");
+
+ assertEquals(1, sentRecords.size());
+ final ObjectNode json = (ObjectNode) parseJson(sentRecords.getFirst().getValue()).get("updatedAttributes");
+ assertEquals("abc", json.get("short").asText(), "Values within limit must not be modified");
+ assertEquals(50, json.get("long").asText().length(), "Values exceeding limit must be trimmed to max length");
+ }
+
+ @Test
+ void testAttributeMaxLengthAtGlobalMaxDisablesTrimming() throws Exception {
+ final String longValue = "x".repeat(100);
+ final Map attrs = new HashMap<>();
+ attrs.put("attr", longValue);
+ final ProvenanceEventRecord event = createEventWithAttributes(attrs, attrs);
+
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.ATTRIBUTE_MAX_LENGTH,
+ String.valueOf(KafkaProvenanceReportingTask.NIFI_DEFAULT_MAX_ATTR_LENGTH));
+
+ final KafkaProvenanceReportingTask task = setup(event, props, 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(-1, task.attributeMaxLength,
+ "Effective max length must be -1 (disabled) when configured value equals the NiFi global limit");
+
+ assertEquals(1, sentRecords.size());
+ final ObjectNode json = (ObjectNode) parseJson(sentRecords.getFirst().getValue()).get("updatedAttributes");
+ assertEquals(longValue, json.get("attr").asText(),
+ "Attribute value must not be trimmed when max length is at the NiFi global limit");
+ }
+
+ @Test
+ void testAttributeMaxLengthAboveGlobalMaxDisablesTrimming() throws Exception {
+ final String longValue = "y".repeat(100);
+ final Map attrs = new HashMap<>();
+ attrs.put("attr", longValue);
+ final ProvenanceEventRecord event = createEventWithAttributes(attrs, attrs);
+
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.ATTRIBUTE_MAX_LENGTH,
+ String.valueOf(KafkaProvenanceReportingTask.NIFI_DEFAULT_MAX_ATTR_LENGTH + 1));
+
+ final KafkaProvenanceReportingTask task = setup(event, props, 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(-1, task.attributeMaxLength,
+ "Effective max length must be -1 (disabled) when configured value exceeds the NiFi global limit");
+
+ assertEquals(1, sentRecords.size());
+ final ObjectNode json = (ObjectNode) parseJson(sentRecords.getFirst().getValue()).get("updatedAttributes");
+ assertEquals(longValue, json.get("attr").asText(),
+ "Attribute value must not be trimmed when max length exceeds the NiFi global limit");
+ }
+
+ @Test
+ void testAttributeMaxLengthBelowMinimumEnforcesFloorOf36() throws Exception {
+ final String value = "a".repeat(40);
+ final Map attrs = new HashMap<>();
+ attrs.put("attr", value);
+ final ProvenanceEventRecord event = createEventWithAttributes(attrs, attrs);
+
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.ATTRIBUTE_MAX_LENGTH, "10");
+
+ final KafkaProvenanceReportingTask task = setup(event, props, 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(KafkaProvenanceReportingTask.ATTR_LENGTH_MIN, task.attributeMaxLength,
+ "Effective max length must be enforced to " + KafkaProvenanceReportingTask.ATTR_LENGTH_MIN
+ + " when configured value is below the minimum");
+
+ assertEquals(1, sentRecords.size());
+ final ObjectNode json = (ObjectNode) parseJson(sentRecords.getFirst().getValue()).get("updatedAttributes");
+ assertEquals(KafkaProvenanceReportingTask.ATTR_LENGTH_MIN, json.get("attr").asText().length(),
+ "Attribute value must be trimmed to the enforced minimum of "
+ + KafkaProvenanceReportingTask.ATTR_LENGTH_MIN + " characters");
+ }
+
+ @Test
+ void testAttributeMaxLengthCombinedWithWhitelistFilter() throws Exception {
+ final Map attrs = new HashMap<>();
+ attrs.put("include", "a".repeat(200));
+ attrs.put("exclude", "b".repeat(200));
+ final ProvenanceEventRecord event = createEventWithAttributes(attrs, attrs);
+
+ final Map props = defaultProperties();
+ props.put(KafkaProvenanceReportingTask.FILTER_ATTRIBUTES_INCLUDE, "include");
+ props.put(KafkaProvenanceReportingTask.ATTRIBUTE_MAX_LENGTH, "50");
+
+ final KafkaProvenanceReportingTask task = setup(event, props, 1);
+ task.initialize(initContext);
+ task.onScheduled(confContext);
+ task.onTrigger(context);
+
+ assertEquals(1, sentRecords.size());
+ final ObjectNode json = (ObjectNode) parseJson(sentRecords.getFirst().getValue()).get("updatedAttributes");
+ assertTrue(json.has("include"), "Whitelisted attribute must be present");
+ assertFalse(json.has("exclude"), "Non-whitelisted attribute must be absent");
+ assertEquals(50, json.get("include").asText().length(),
+ "Whitelisted attribute value must be trimmed to max length");
+ }
+}
diff --git a/nifi-extension-bundles/nifi-kafka-bundle/pom.xml b/nifi-extension-bundles/nifi-kafka-bundle/pom.xml
index 857d34d730ef..60f140479a60 100644
--- a/nifi-extension-bundles/nifi-kafka-bundle/pom.xml
+++ b/nifi-extension-bundles/nifi-kafka-bundle/pom.xml
@@ -41,6 +41,7 @@
nifi-kafka-service-aws-nar
nifi-kafka-service-shared
nifi-kafka-shared
+ nifi-kafka-provenance-reporting-task