This is an automated email from the ASF dual-hosted git repository.
pvillard31 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 0ef04bc95c9 NIFI-15979 Wire Connector MDC and reporting-task
visibility (#11330)
0ef04bc95c9 is described below
commit 0ef04bc95c9b3946d51bc76d948250f2d34f3318
Author: Kevin Doran <[email protected]>
AuthorDate: Fri Jun 12 23:19:09 2026 -0400
NIFI-15979 Wire Connector MDC and reporting-task visibility (#11330)
Surface Connector observability that was part of nifi-api 2.9.0:
- ConnectorNode.getLoggingAttributes() plus
StandardConnectorNode/StandardProcessGroup
propagation of connector identity into descendant-component MDC.
- StandardEventAccess.getConnectorStatuses() returns a ConnectorStatus (id,
name, root
group status) per connector; StatusHistoryRepository gains a capture
overload accepting
Collection<ConnectorStatus>; FlowController invokes it.
---
.../src/main/asciidoc/administration-guide.adoc | 16 ++-
.../connector/ConnectorConfigurationProvider.java | 29 ++++++
.../status/history/StatusHistoryRepository.java | 21 ++++
.../EmbeddedQuestDbStatusHistoryRepository.java | 27 +++++
.../AbstractStatusHistoryRepositoryTest.java | 21 ++++
...DbStatusHistoryRepositoryForComponentsTest.java | 19 ++++
.../apache/nifi/groups/StandardProcessGroup.java | 48 +++++++++
.../nifi/groups/StandardProcessGroupTest.java | 108 ++++++++++++++++++++
.../nifi/components/connector/ConnectorNode.java | 25 +++++
.../java/org/apache/nifi/groups/ProcessGroup.java | 22 ++++
.../connector/ConnectorLoggingAttribute.java | 62 ++++++++++++
.../connector/StandardConnectorNode.java | 112 ++++++++++++++++++++-
.../connector/StandardConnectorRepository.java | 42 ++++++++
.../apache/nifi/controller/ExtensionBuilder.java | 16 ++-
.../org/apache/nifi/controller/FlowController.java | 5 +-
.../history/VolatileComponentStatusRepository.java | 20 ++++
.../apache/nifi/reporting/StandardEventAccess.java | 53 +++++++++-
.../connector/TestStandardConnectorNode.java | 82 +++++++++++++++
.../connector/TestStandardConnectorRepository.java | 102 +++++++++++++++++++
.../AbstractStatusHistoryRepositoryTest.java | 21 ++++
...latileComponentStatusRepositoryForNodeTest.java | 29 ++++++
21 files changed, 871 insertions(+), 9 deletions(-)
diff --git a/nifi-docs/src/main/asciidoc/administration-guide.adoc
b/nifi-docs/src/main/asciidoc/administration-guide.adoc
index d5e4234bc39..f7ed3cf5702 100644
--- a/nifi-docs/src/main/asciidoc/administration-guide.adoc
+++ b/nifi-docs/src/main/asciidoc/administration-guide.adoc
@@ -211,11 +211,25 @@ Component logs provide the following MDC named values:
- `processGroupName` contains the name of the Process Group
- `processGroupNamePath` contains of the hierarchy of names for Process Groups
with separators
+Components that run inside a Connector-managed flow also carry
framework-supplied MDC values that identify the owning
+Connector:
+
+- `connectorId` contains the UUID of the Connector
+- `connectorName` contains the user-visible name of the Connector
+- `connectorComponent` contains the fully qualified class name of the
Connector implementation
+- `connectorBundleGroup`, `connectorBundleArtifact`, `connectorBundleVersion`
identify the NAR bundle the Connector was
+ loaded from
+
+Additional implementation-specific MDC values may be supplied by the framework
+`ConnectorConfigurationProvider` extension via `getLoggingAttributes(String
connectorId)`. The framework consults the
+provider at connector creation time and merges the result with the
framework-managed keys above. Keys reserved by the
+framework (those listed above) cannot be overridden; attempts to do so are
dropped and logged as a `WARN`.
+
MDC named values can be added to a Logback pattern layout using the `mdc`
conversion word.
[source, xml]
----
-<pattern>%date %level [%thread] %mdc{processGroupId} %logger{40}
%msg%n</pattern>
+<pattern>%date %level [%thread] %mdc{connectorId} %mdc{processGroupId}
%logger{40} %msg%n</pattern>
----
Logs from classes other than extension components do not have MDC named
values. Logs formatted using the pattern layout
diff --git
a/nifi-framework-api/src/main/java/org/apache/nifi/components/connector/ConnectorConfigurationProvider.java
b/nifi-framework-api/src/main/java/org/apache/nifi/components/connector/ConnectorConfigurationProvider.java
index a6327b2a8ba..16e5ae6e9b2 100644
---
a/nifi-framework-api/src/main/java/org/apache/nifi/components/connector/ConnectorConfigurationProvider.java
+++
b/nifi-framework-api/src/main/java/org/apache/nifi/components/connector/ConnectorConfigurationProvider.java
@@ -20,6 +20,7 @@ package org.apache.nifi.components.connector;
import org.apache.nifi.flow.VersionedConnectorState;
import java.io.InputStream;
+import java.util.Map;
import java.util.Optional;
/**
@@ -212,6 +213,34 @@ public interface ConnectorConfigurationProvider {
return true;
}
+ /**
+ * Returns identity-shaped logging attributes the framework should include
in the SLF4J {@code MDC} for
+ * every log line emitted by the Connector with the given identifier and
by any component (Processor,
+ * Controller Service, etc.) running inside its managed flow. The same
attributes are also snapshotted
+ * onto {@code ProcessGroupStatus} so that status-based reporting tasks
(e.g. OpenTelemetry) can attach
+ * them as metric attributes.
+ *
+ * <p>The framework reserves a set of well-known keys that describe the
Connector itself (identity,
+ * bundle coordinate, etc.). Any entry in the returned map whose key
collides with a reserved
+ * framework-managed key is dropped by the framework and a {@code WARN} is
logged for that entry;
+ * the remaining entries are accepted.</p>
+ *
+ * <p>The framework consults this method once, at connector create time.
It is the provider's
+ * responsibility to ensure the returned values are stable for the
lifetime of the connector
+ * instance or to publish updates through other means; mid-lifecycle
refresh is not currently
+ * defined.</p>
+ *
+ * <p>The default returns {@link Map#of()} so that providers that do not
need to publish additional
+ * attributes (and runtimes that have no provider configured) get only the
framework-managed identity
+ * keys in MDC.</p>
+ *
+ * @param connectorId the identifier of the connector whose logging
attributes are being requested
+ * @return an immutable map of attribute keys to values; never {@code null}
+ */
+ default Map<String, String> getLoggingAttributes(final String connectorId)
{
+ return Map.of();
+ }
+
/**
* Ensures that local asset binaries are up to date with the external
store. For each asset
* tracked in the provider's local state, this method compares the
external store's current
diff --git
a/nifi-framework-api/src/main/java/org/apache/nifi/controller/status/history/StatusHistoryRepository.java
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/status/history/StatusHistoryRepository.java
index 2b96feb6192..b0225855281 100644
---
a/nifi-framework-api/src/main/java/org/apache/nifi/controller/status/history/StatusHistoryRepository.java
+++
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/status/history/StatusHistoryRepository.java
@@ -16,9 +16,11 @@
*/
package org.apache.nifi.controller.status.history;
+import org.apache.nifi.controller.status.ConnectorStatus;
import org.apache.nifi.controller.status.NodeStatus;
import org.apache.nifi.controller.status.ProcessGroupStatus;
+import java.util.Collection;
import java.util.Date;
import java.util.List;
@@ -48,6 +50,25 @@ public interface StatusHistoryRepository {
*/
void capture(NodeStatus nodeStatus, ProcessGroupStatus rootGroupStatus,
List<GarbageCollectionStatus> garbageCollectionStatus, Date timestamp);
+ /**
+ * Captures the status information provided in the given report,
additionally providing the status of each
+ * Connector-managed flow. Connector-managed Process Groups are siblings
of the root group and are not reachable
+ * from {@code rootGroupStatus}, so this overload allows implementations
to observe them (for example, to export
+ * Connector-managed flow metrics). The default implementation ignores
{@code connectorStatuses} and delegates to
+ * {@link #capture(NodeStatus, ProcessGroupStatus, List, Date)} so
existing implementations continue to work
+ * unchanged.
+ *
+ * @param nodeStatus status of the node
+ * @param rootGroupStatus status of the root group and its content
+ * @param connectorStatuses status of each registered Connector, each
carrying its managed root group status
+ * @param garbageCollectionStatus status of garbage collection
+ * @param timestamp timestamp of capture
+ */
+ default void capture(NodeStatus nodeStatus, ProcessGroupStatus
rootGroupStatus, Collection<ConnectorStatus> connectorStatuses,
+ List<GarbageCollectionStatus>
garbageCollectionStatus, Date timestamp) {
+ capture(nodeStatus, rootGroupStatus, garbageCollectionStatus,
timestamp);
+ }
+
/**
* @param connectionId the ID of the Connection for which the Status is
* desired
diff --git
a/nifi-framework-bundle/nifi-framework-extensions/nifi-questdb-bundle/nifi-questdb-status-history/src/main/java/org/apache/nifi/controller/status/history/questdb/EmbeddedQuestDbStatusHistoryRepository.java
b/nifi-framework-bundle/nifi-framework-extensions/nifi-questdb-bundle/nifi-questdb-status-history/src/main/java/org/apache/nifi/controller/status/history/questdb/EmbeddedQuestDbStatusHistoryRepository.java
index 81966b606e2..4afd292ddd3 100644
---
a/nifi-framework-bundle/nifi-framework-extensions/nifi-questdb-bundle/nifi-questdb-status-history/src/main/java/org/apache/nifi/controller/status/history/questdb/EmbeddedQuestDbStatusHistoryRepository.java
+++
b/nifi-framework-bundle/nifi-framework-extensions/nifi-questdb-bundle/nifi-questdb-status-history/src/main/java/org/apache/nifi/controller/status/history/questdb/EmbeddedQuestDbStatusHistoryRepository.java
@@ -17,6 +17,7 @@
package org.apache.nifi.controller.status.history.questdb;
import org.apache.nifi.controller.status.ConnectionStatus;
+import org.apache.nifi.controller.status.ConnectorStatus;
import org.apache.nifi.controller.status.NodeStatus;
import org.apache.nifi.controller.status.ProcessGroupStatus;
import org.apache.nifi.controller.status.ProcessorStatus;
@@ -126,6 +127,32 @@ public class EmbeddedQuestDbStatusHistoryRepository
implements StatusHistoryRepo
updateComponentDetails(rootGroupStatus);
}
+ @Override
+ public void capture(final NodeStatus nodeStatus, final ProcessGroupStatus
rootGroupStatus, final Collection<ConnectorStatus> connectorStatuses,
+ final List<GarbageCollectionStatus>
garbageCollectionStatus, final Date timestamp) {
+ final Instant captured = timestamp.toInstant();
+ captureNodeStatus(nodeStatus, captured);
+ captureGarbageCollectionStatus(garbageCollectionStatus, captured);
+ captureComponentStatus(rootGroupStatus, captured);
+
+ // Connector-managed flows are siblings of the root group and
therefore are not reachable from
+ // rootGroupStatus. Capture each Connector's managed flow so that
components running inside it have
+ // status history just like components in the primary flow. Component
details for the root group and
+ // every connector-managed flow are accumulated into a single map so
that one does not overwrite another.
+ final Map<String, ComponentDetails> componentDetails = new HashMap<>();
+ updateComponentDetails(rootGroupStatus, componentDetails);
+ if (connectorStatuses != null) {
+ for (final ConnectorStatus connectorStatus : connectorStatuses) {
+ final ProcessGroupStatus connectorRootGroupStatus =
connectorStatus.getRootGroupStatus();
+ if (connectorRootGroupStatus != null) {
+ captureComponentStatus(connectorRootGroupStatus, captured);
+ updateComponentDetails(connectorRootGroupStatus,
componentDetails);
+ }
+ }
+ }
+ componentDetailsProvider.setComponentDetails(componentDetails);
+ }
+
private void captureNodeStatus(final NodeStatus nodeStatus, final Instant
captured) {
storage.storeNodeStatuses(Collections.singleton(new
CapturedStatus<>(nodeStatus, captured)));
}
diff --git
a/nifi-framework-bundle/nifi-framework-extensions/nifi-questdb-bundle/nifi-questdb-status-history/src/test/java/org/apache/nifi/controller/status/history/AbstractStatusHistoryRepositoryTest.java
b/nifi-framework-bundle/nifi-framework-extensions/nifi-questdb-bundle/nifi-questdb-status-history/src/test/java/org/apache/nifi/controller/status/history/AbstractStatusHistoryRepositoryTest.java
index 62f865721c9..2b032e62b89 100644
---
a/nifi-framework-bundle/nifi-framework-extensions/nifi-questdb-bundle/nifi-questdb-status-history/src/test/java/org/apache/nifi/controller/status/history/AbstractStatusHistoryRepositoryTest.java
+++
b/nifi-framework-bundle/nifi-framework-extensions/nifi-questdb-bundle/nifi-questdb-status-history/src/test/java/org/apache/nifi/controller/status/history/AbstractStatusHistoryRepositoryTest.java
@@ -17,6 +17,7 @@
package org.apache.nifi.controller.status.history;
import org.apache.nifi.controller.status.ConnectionStatus;
+import org.apache.nifi.controller.status.ConnectorStatus;
import org.apache.nifi.controller.status.NodeStatus;
import org.apache.nifi.controller.status.ProcessGroupStatus;
import org.apache.nifi.controller.status.ProcessorStatus;
@@ -41,6 +42,9 @@ public abstract class AbstractStatusHistoryRepositoryTest {
protected static final String PROCESSOR_WITH_COUNTER_ID =
"f09c5bb8-7b67-4d3b-81c5-5373e1c03ed1";
protected static final String CONNECTION_ID =
"d5452f70-a0d9-44c5-aeda-c2026482e4ee";
protected static final String REMOTE_PROCESS_GROUP_ID =
"f5f4fe2a-0209-4ba7-8f15-f33df942cde5";
+ protected static final String CONNECTOR_ID =
"a1111111-1111-1111-1111-111111111111";
+ protected static final String CONNECTOR_ROOT_GROUP_ID =
"a2222222-2222-2222-2222-222222222222";
+ protected static final String CONNECTOR_PROCESSOR_ID =
"a3333333-3333-3333-3333-333333333333";
protected ProcessGroupStatus givenSimpleRootProcessGroupStatus() {
final ProcessGroupStatus status = new ProcessGroupStatus();
@@ -74,6 +78,23 @@ public abstract class AbstractStatusHistoryRepositoryTest {
return status;
}
+ protected ConnectorStatus givenConnectorStatus() {
+ final ProcessorStatus connectorProcessor = givenProcessorStatus();
+ connectorProcessor.setId(CONNECTOR_PROCESSOR_ID);
+ connectorProcessor.setName("ConnectorProcessor");
+
+ final ProcessGroupStatus connectorRootGroup =
givenSimpleRootProcessGroupStatus();
+ connectorRootGroup.setId(CONNECTOR_ROOT_GROUP_ID);
+ connectorRootGroup.setName("Connector Root");
+
connectorRootGroup.setProcessorStatus(Collections.singleton(connectorProcessor));
+
+ final ConnectorStatus connectorStatus = new ConnectorStatus();
+ connectorStatus.setId(CONNECTOR_ID);
+ connectorStatus.setName("Test Connector");
+ connectorStatus.setRootGroupStatus(connectorRootGroup);
+ return connectorStatus;
+ }
+
protected ProcessGroupStatus givenChildProcessGroupStatus() {
final ProcessGroupStatus status = new ProcessGroupStatus();
status.setId(CHILD_GROUP_ID);
diff --git
a/nifi-framework-bundle/nifi-framework-extensions/nifi-questdb-bundle/nifi-questdb-status-history/src/test/java/org/apache/nifi/controller/status/history/EmbeddedQuestDbStatusHistoryRepositoryForComponentsTest.java
b/nifi-framework-bundle/nifi-framework-extensions/nifi-questdb-bundle/nifi-questdb-status-history/src/test/java/org/apache/nifi/controller/status/history/EmbeddedQuestDbStatusHistoryRepositoryForComponentsTest.java
index e9e9db41b6d..95d1ce5096e 100644
---
a/nifi-framework-bundle/nifi-framework-extensions/nifi-questdb-bundle/nifi-questdb-status-history/src/test/java/org/apache/nifi/controller/status/history/EmbeddedQuestDbStatusHistoryRepositoryForComponentsTest.java
+++
b/nifi-framework-bundle/nifi-framework-extensions/nifi-questdb-bundle/nifi-questdb-status-history/src/test/java/org/apache/nifi/controller/status/history/EmbeddedQuestDbStatusHistoryRepositoryForComponentsTest.java
@@ -21,6 +21,7 @@ import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.Date;
+import java.util.List;
import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -79,6 +80,24 @@ public class
EmbeddedQuestDbStatusHistoryRepositoryForComponentsTest extends Abs
assertStatusHistoryIsEmpty(rootGroupStatus2);
}
+ @Test
+ public void testWritingThenReadingConnectorManagedFlowComponents() throws
Exception {
+ repository.capture(new NodeStatus(), givenRootProcessGroupStatus(),
List.of(givenConnectorStatus()), new ArrayList<>(), INSERTED_AT);
+ waitUntilPersisted();
+
+ // The Connector's managed root group is captured even though it is a
sibling of (not reachable from) the root group
+ final StatusHistory connectorRootGroupStatus =
repository.getProcessGroupStatusHistory(CONNECTOR_ROOT_GROUP_ID, START, END,
PREFERRED_DATA_POINTS);
+ assertCorrectStatusHistory(connectorRootGroupStatus,
CONNECTOR_ROOT_GROUP_ID, "Connector Root");
+
+ // A processor running inside the Connector-managed flow has status
history just like a primary-flow processor
+ final StatusHistory connectorProcessorStatus =
repository.getProcessorStatusHistory(CONNECTOR_PROCESSOR_ID, START, END,
PREFERRED_DATA_POINTS, false);
+ assertCorrectStatusHistory(connectorProcessorStatus,
CONNECTOR_PROCESSOR_ID, "ConnectorProcessor");
+
+ // The primary root flow is still captured alongside the connector flow
+ final StatusHistory rootGroupStatus =
repository.getProcessGroupStatusHistory(ROOT_GROUP_ID, START, END,
PREFERRED_DATA_POINTS);
+ assertCorrectStatusHistory(rootGroupStatus, ROOT_GROUP_ID, "Root");
+ }
+
@Test
public void testReadingLimitedByPreferredDataPoints() throws Exception {
repository.capture(new NodeStatus(),
givenSimpleRootProcessGroupStatus(), new ArrayList<>(), new
Date(INSERTED_AT.getTime() - TimeUnit.MINUTES.toMillis(8)));
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
index 91f3bdbde4e..610eb3f6413 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
@@ -234,6 +234,7 @@ public final class StandardProcessGroup implements
ProcessGroup {
private static final String UNREGISTERED_PATH_SEGMENT = "UNREGISTERED";
private final Map<String, String> loggingAttributes = new
ConcurrentHashMap<>();
+ private volatile Map<String, String> connectorLoggingAttributes = Map.of();
private volatile String logFileSuffix;
public StandardProcessGroup(final String id, final
ControllerServiceProvider serviceProvider, final ProcessScheduler scheduler,
@@ -308,6 +309,12 @@ public final class StandardProcessGroup implements
ProcessGroup {
@Override
public void setParent(final ProcessGroup newParent) {
parent.set(newParent);
+ // Inherit connector-supplied MDC attributes from the parent so
descendants of a connector's managed
+ // flow carry the same connector metadata (attributing their logs and
status metrics to the connector).
+ // Runs on every re-parent (including initial attach), so PGs added
later inherit automatically.
+ if (newParent instanceof StandardProcessGroup standardParent) {
+ this.connectorLoggingAttributes =
standardParent.connectorLoggingAttributes;
+ }
setLoggingAttributes();
}
@@ -4829,6 +4836,47 @@ public final class StandardProcessGroup implements
ProcessGroup {
final String registeredFlowVersion =
currentVersionControl.getVersion();
loggingAttributes.put(LoggingAttribute.REGISTERED_FLOW_VERSION.attribute,
registeredFlowVersion);
}
+
+ loggingAttributes.putAll(connectorLoggingAttributes);
+ }
+
+ /**
+ * Stores the connector-managed MDC attributes for this process group and
cascades the same
+ * attributes to all descendant process groups so that components anywhere
in the connector's
+ * managed flow log with consistent connectorId/connectorName/etc. context.
+ *
+ * <p>This method is called by {@code StandardConnectorNode} against its
managed root process
+ * group whenever the connector's framework keys (e.g. {@code
connectorName}) change or when the
+ * connector provides updated custom logging attributes. Newly created
descendant groups will
+ * also inherit the attributes lazily via {@link
#setParent(ProcessGroup)}.</p>
+ *
+ * @param attributes the merged set of connector logging attributes; an
empty or {@code null}
+ * map clears any previously assigned attributes
+ */
+ @Override
+ public void setConnectorLoggingAttributes(final Map<String, String>
attributes) {
+ final Map<String, String> snapshot = (attributes == null ||
attributes.isEmpty())
+ ? Map.of()
+ : Map.copyOf(attributes);
+ this.connectorLoggingAttributes = snapshot;
+ setLoggingAttributes();
+
+ for (final ProcessGroup child : getProcessGroups()) {
+ if (child instanceof StandardProcessGroup standardChild) {
+ standardChild.setConnectorLoggingAttributes(snapshot);
+ }
+ }
+ }
+
+ /**
+ * Returns the connector-managed MDC attributes currently assigned to this
process group (inherited from
+ * its connector's managed flow root, or empty if this group is not part
of a connector flow).
+ *
+ * @return an immutable map of connector logging attributes; never {@code
null}
+ */
+ @Override
+ public Map<String, String> getConnectorLoggingAttributes() {
+ return connectorLoggingAttributes;
}
private void setGroupPath() {
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/StandardProcessGroupTest.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/StandardProcessGroupTest.java
index 35051a30f9f..592721b466e 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/StandardProcessGroupTest.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/StandardProcessGroupTest.java
@@ -44,6 +44,7 @@ import java.util.Map;
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.assertTrue;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.when;
@@ -238,4 +239,111 @@ class StandardProcessGroupTest {
assertEquals(expected, loggingAttributes);
}
+
+ @Test
+ void testSetConnectorLoggingAttributesMergesIntoLoggingAttributes() {
+ processGroup.setName(NAME);
+
+ final Map<String, String> connectorAttributes = Map.of(
+ "connectorId", "connector-1",
+ "connectorName", "My Connector",
+ "connectorComponent", "com.example.MyConnector"
+ );
+
+ processGroup.setConnectorLoggingAttributes(connectorAttributes);
+
+ final Map<String, String> loggingAttributes =
processGroup.getLoggingAttributes();
+ assertEquals("connector-1", loggingAttributes.get("connectorId"));
+ assertEquals("My Connector", loggingAttributes.get("connectorName"));
+ assertEquals("com.example.MyConnector",
loggingAttributes.get("connectorComponent"));
+ assertEquals(NAME,
loggingAttributes.get(StandardProcessGroup.LoggingAttribute.PROCESS_GROUP_NAME.getAttribute()));
+ assertEquals(ID,
loggingAttributes.get(StandardProcessGroup.LoggingAttribute.PROCESS_GROUP_ID.getAttribute()));
+ }
+
+ @Test
+ void testSetConnectorLoggingAttributesCascadesToChildProcessGroups() {
+ processGroup.setName(NAME);
+
+ final StandardProcessGroup child =
createStandardProcessGroup("child-id");
+ child.setName("Child");
+ processGroup.addProcessGroup(child);
+
+ final Map<String, String> connectorAttributes = Map.of(
+ "connectorId", "connector-1",
+ "connectorName", "Postgres CDC"
+ );
+
+ processGroup.setConnectorLoggingAttributes(connectorAttributes);
+
+ assertEquals("connector-1",
child.getLoggingAttributes().get("connectorId"));
+ assertEquals("Postgres CDC",
child.getLoggingAttributes().get("connectorName"));
+ }
+
+ @Test
+ void testAddProcessGroupInheritsConnectorLoggingAttributesFromParent() {
+ processGroup.setName(NAME);
+ processGroup.setConnectorLoggingAttributes(Map.of(
+ "connectorId", "connector-1",
+ "connectorName", "Postgres CDC"
+ ));
+
+ final StandardProcessGroup lateChild =
createStandardProcessGroup("late-child-id");
+ lateChild.setName("Late Child");
+ processGroup.addProcessGroup(lateChild);
+
+ assertEquals("connector-1",
lateChild.getLoggingAttributes().get("connectorId"));
+ assertEquals("Postgres CDC",
lateChild.getLoggingAttributes().get("connectorName"));
+ assertEquals(Map.copyOf(processGroup.getConnectorLoggingAttributes()),
lateChild.getConnectorLoggingAttributes());
+ }
+
+ @Test
+ void testEmptyConnectorLoggingAttributesAddsNothing() {
+ processGroup.setName(NAME);
+ processGroup.setConnectorLoggingAttributes(Map.of());
+
+ final Map<String, String> loggingAttributes =
processGroup.getLoggingAttributes();
+ assertFalse(loggingAttributes.containsKey("connectorId"));
+ assertTrue(processGroup.getConnectorLoggingAttributes().isEmpty());
+ // Verify the PG-level keys are still present.
+ assertEquals(NAME,
loggingAttributes.get(StandardProcessGroup.LoggingAttribute.PROCESS_GROUP_NAME.getAttribute()));
+ }
+
+ @Test
+ void testSetConnectorLoggingAttributesReplacesPreviousValues() {
+ processGroup.setName(NAME);
+ processGroup.setConnectorLoggingAttributes(Map.of(
+ "connectorId", "connector-1",
+ "connectorName", "Old Name",
+ "customKey", "customValue"
+ ));
+
+ processGroup.setConnectorLoggingAttributes(Map.of(
+ "connectorId", "connector-1",
+ "connectorName", "New Name"
+ ));
+
+ final Map<String, String> loggingAttributes =
processGroup.getLoggingAttributes();
+ assertEquals("New Name", loggingAttributes.get("connectorName"));
+ assertFalse(loggingAttributes.containsKey("customKey"));
+ }
+
+ private StandardProcessGroup createStandardProcessGroup(final String id) {
+ return new StandardProcessGroup(
+ id,
+ controllerServiceProvider,
+ processScheduler,
+ propertyEncryptor,
+ extensionManager,
+ stateManagerProvider,
+ flowManager,
+ reloadComponent,
+ nodeTypeProvider,
+ clusterTopologyProvider,
+ properties,
+ statelessGroupNodeFactory,
+ assetManager,
+ null
+ );
+ }
+
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorNode.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorNode.java
index 1c3c78073b6..fd994aa8e79 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorNode.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorNode.java
@@ -35,6 +35,7 @@ import org.apache.nifi.logging.ComponentLog;
import java.time.Duration;
import java.util.Collection;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.concurrent.Future;
@@ -67,6 +68,30 @@ public interface ConnectorNode extends
ComponentAuthorizable, VersionedComponent
BundleCoordinate getBundleCoordinate();
+ /**
+ * @return an immutable map of the Connector's logging attributes: the
framework-managed identity keys
+ * (such as connector identifier, name, component type, and bundle
coordinate) merged with any
+ * provider-supplied custom attributes. These are the attributes
applied to the MDC of the Connector's
+ * managed flow and surfaced on Connector metrics. Never {@code
null}. The default implementation
+ * returns an empty map.
+ */
+ default Map<String, String> getLoggingAttributes() {
+ return Map.of();
+ }
+
+ /**
+ * Replaces the provider-supplied custom logging attributes for this
Connector, merging them with the
+ * framework-managed identity keys (see {@link #getLoggingAttributes()}).
This is invoked by the framework
+ * when the Connector node is added or restored, using attributes sourced
from the
+ * {@link ConnectorConfigurationProvider} (if one is configured). Keys
reserved by the framework are ignored.
+ * The default implementation is a no-op.
+ *
+ * @param attributes the proposed custom attributes; {@code null} or empty
clears any previously supplied
+ * custom attributes (the framework-managed keys remain)
+ */
+ default void setCustomLoggingAttributes(final Map<String, String>
attributes) {
+ }
+
/**
* Returns whether or not the underlying extension is missing (i.e., the
Connector is a GhostConnector).
* @return true if the extension is missing, false otherwise
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/ProcessGroup.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/ProcessGroup.java
index 25bf1cd2b22..976004b514b 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/ProcessGroup.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/ProcessGroup.java
@@ -1235,6 +1235,28 @@ public interface ProcessGroup extends
ComponentAuthorizable, Positionable, Versi
*/
Map<String, String> getLoggingAttributes();
+ /**
+ * Returns the connector-managed MDC attributes assigned to this Process
Group. When this group is part of a
+ * Connector's managed flow, these attributes (for example, connectorId
and connectorName) are inherited from the
+ * Connector's managed root group so that logs and status metrics emitted
by components in the flow can be
+ * attributed to the Connector. Returns an empty map when this group is
not part of a Connector flow.
+ *
+ * @return an immutable map of connector logging attributes; never {@code
null}
+ */
+ default Map<String, String> getConnectorLoggingAttributes() {
+ return Map.of();
+ }
+
+ /**
+ * Assigns the connector-managed MDC attributes for this Process Group and
cascades them to all descendant Process
+ * Groups, so that components anywhere in a Connector's managed flow log
with consistent connector context. An empty
+ * or {@code null} map clears any previously assigned attributes.
+ *
+ * @param attributes the connector logging attributes to assign
+ */
+ default void setConnectorLoggingAttributes(final Map<String, String>
attributes) {
+ }
+
/**
* @return the log file suffix of the ProcessGroup for dedicated logging
*/
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/ConnectorLoggingAttribute.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/ConnectorLoggingAttribute.java
new file mode 100644
index 00000000000..46d936f8ded
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/ConnectorLoggingAttribute.java
@@ -0,0 +1,62 @@
+/*
+ * 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.components.connector;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Framework-managed MDC logging attribute keys for a connector. The framework
computes these
+ * automatically and they cannot be overridden by a connector's custom
attributes.
+ */
+public enum ConnectorLoggingAttribute {
+ CONNECTOR_ID("connectorId"),
+ CONNECTOR_NAME("connectorName"),
+ CONNECTOR_COMPONENT("connectorComponent"),
+ CONNECTOR_BUNDLE_GROUP("connectorBundleGroup"),
+ CONNECTOR_BUNDLE_ARTIFACT("connectorBundleArtifact"),
+ CONNECTOR_BUNDLE_VERSION("connectorBundleVersion");
+
+ private static final Set<String> RESERVED_KEYS;
+ static {
+ final Set<String> reserved = new HashSet<>();
+ for (final ConnectorLoggingAttribute attribute : values()) {
+ reserved.add(attribute.attribute);
+ }
+ RESERVED_KEYS = Collections.unmodifiableSet(reserved);
+ }
+
+ private final String attribute;
+
+ ConnectorLoggingAttribute(final String attribute) {
+ this.attribute = attribute;
+ }
+
+ public String getAttribute() {
+ return attribute;
+ }
+
+ public static boolean isReserved(final String key) {
+ return RESERVED_KEYS.contains(key);
+ }
+
+ public static Set<String> getReservedKeys() {
+ return RESERVED_KEYS;
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java
index 8f5ae6f123a..999471373b6 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java
@@ -53,6 +53,7 @@ import org.apache.nifi.flow.VersionedProcessor;
import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.groups.RemoteProcessGroup;
import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.logging.GroupedComponent;
import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.nar.NarCloseable;
import org.apache.nifi.util.StringUtils;
@@ -84,7 +85,7 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.stream.Collectors;
-public class StandardConnectorNode implements ConnectorNode {
+public class StandardConnectorNode implements ConnectorNode, GroupedComponent {
private static final Logger logger =
LoggerFactory.getLogger(StandardConnectorNode.class);
private static final Set<org.apache.nifi.controller.ScheduledState>
STOPPED_STATES =
@@ -115,6 +116,10 @@ public class StandardConnectorNode implements
ConnectorNode {
private volatile String name;
private volatile FrameworkConnectorInitializationContext
initializationContext;
+ private final Object loggingAttributesLock = new Object();
+ private volatile Map<String, String> customLoggingAttributes = Map.of();
+ private volatile Map<String, String> mergedLoggingAttributes = Map.of();
+
public StandardConnectorNode(final String identifier, final FlowManager
flowManager, final ExtensionManager extensionManager,
final Authorizable parentAuthorizable, final ConnectorDetails
connectorDetails, final String componentType, final String
componentCanonicalClass,
@@ -139,6 +144,8 @@ public class StandardConnectorNode implements ConnectorNode
{
final Bundle activeFlowBundle = new
Bundle(bundleCoordinate.getGroup(), bundleCoordinate.getId(),
bundleCoordinate.getVersion());
this.activeFlowContext =
flowContextFactory.createActiveFlowContext(identifier,
connectorDetails.getComponentLog(), activeFlowBundle);
+
+ rebuildLoggingAttributes();
}
@Override
@@ -149,6 +156,109 @@ public class StandardConnectorNode implements
ConnectorNode {
@Override
public void setName(final String name) {
this.name = name;
+ rebuildLoggingAttributes();
+ }
+
+ /**
+ * Returns the {@link ProcessGroup} that holds the connector-managed flow
(the connector's active
+ * managed root group), or {@code null} before the active flow context has
been established.
+ *
+ * <p>This serves two purposes:
+ * <ul>
+ * <li>It satisfies the {@link GroupedComponent} contract. The
connector's own {@code ComponentLog}
+ * uses a {@code StandardLoggingContext} bound to this node (see
{@code ExtensionBuilder}), so the
+ * connector's own log lines source their MDC from this group's
+ * {@link ProcessGroup#getLoggingAttributes() logging
attributes}.</li>
+ * <li>It is the target onto which {@link #rebuildLoggingAttributes()}
pushes the merged connector
+ * logging attributes (via {@link
ProcessGroup#setConnectorLoggingAttributes(Map)}); those then
+ * cascade down the managed flow so components running inside it
inherit the connector MDC through
+ * their own logging contexts.</li>
+ * </ul>
+ *
+ * @return the connector's managed root process group, or {@code null}
before the active flow context
+ * has been established
+ */
+ @Override
+ public ProcessGroup getProcessGroup() {
+ final FrameworkFlowContext context = activeFlowContext;
+ return context == null ? null : context.getManagedProcessGroup();
+ }
+
+ /**
+ * Replaces the connector-supplied custom logging attributes. Reserved
keys (those used by the
+ * framework, see {@link ConnectorLoggingAttribute}) are filtered out and
a WARN is logged for
+ * each dropped entry.
+ *
+ * @param attributes the proposed custom attributes; {@code null} or empty
clears the current set
+ */
+ @Override
+ public void setCustomLoggingAttributes(final Map<String, String>
attributes) {
+ final Map<String, String> filtered = filterReservedKeys(attributes);
+
+ synchronized (loggingAttributesLock) {
+ this.customLoggingAttributes = filtered;
+ }
+ rebuildLoggingAttributes();
+ }
+
+ /**
+ * Returns an immutable snapshot of the merged framework + custom logging
attributes currently
+ * advertised by this connector. The framework keys are populated by the
framework from the
+ * connector's identifier, name, component type, and bundle coordinate.
+ *
+ * @return an immutable map of the connector's merged framework and custom
logging attributes; never {@code null}
+ */
+ @Override
+ public Map<String, String> getLoggingAttributes() {
+ return mergedLoggingAttributes;
+ }
+
+ private Map<String, String> filterReservedKeys(final Map<String, String>
attributes) {
+ if (attributes == null || attributes.isEmpty()) {
+ return Map.of();
+ }
+ final Map<String, String> filtered = new HashMap<>(attributes.size());
+ for (final Map.Entry<String, String> entry : attributes.entrySet()) {
+ final String key = entry.getKey();
+ if (key == null || key.isEmpty()) {
+ continue;
+ }
+ if (ConnectorLoggingAttribute.isReserved(key)) {
+ logger.warn("{} attempted to set reserved logging attribute
[{}]; dropping the entry. Reserved keys are managed by the framework.", this,
key);
+ continue;
+ }
+ filtered.put(key, entry.getValue());
+ }
+ return Collections.unmodifiableMap(filtered);
+ }
+
+ private void rebuildLoggingAttributes() {
+ final Map<String, String> merged;
+ final Map<String, String> custom;
+ synchronized (loggingAttributesLock) {
+ custom = customLoggingAttributes;
+ merged = new HashMap<>(custom.size() +
ConnectorLoggingAttribute.values().length);
+ merged.putAll(custom);
+ // Framework keys are applied last so they always win against any
not-yet-filtered overlap.
+ merged.put(ConnectorLoggingAttribute.CONNECTOR_ID.getAttribute(),
identifier);
+
merged.put(ConnectorLoggingAttribute.CONNECTOR_NAME.getAttribute(), name ==
null ? "" : name);
+
merged.put(ConnectorLoggingAttribute.CONNECTOR_COMPONENT.getAttribute(),
componentCanonicalClass == null ? "" : componentCanonicalClass);
+ if (bundleCoordinate != null) {
+
merged.put(ConnectorLoggingAttribute.CONNECTOR_BUNDLE_GROUP.getAttribute(),
bundleCoordinate.getGroup());
+
merged.put(ConnectorLoggingAttribute.CONNECTOR_BUNDLE_ARTIFACT.getAttribute(),
bundleCoordinate.getId());
+
merged.put(ConnectorLoggingAttribute.CONNECTOR_BUNDLE_VERSION.getAttribute(),
bundleCoordinate.getVersion());
+ }
+ this.mergedLoggingAttributes = Collections.unmodifiableMap(merged);
+ }
+
+ pushLoggingAttributesToManagedFlow(merged);
+ }
+
+ private void pushLoggingAttributesToManagedFlow(final Map<String, String>
attributes) {
+ final ProcessGroup managedProcessGroup = getProcessGroup();
+ if (managedProcessGroup != null) {
+ managedProcessGroup.setConnectorLoggingAttributes(attributes);
+ }
}
@Override
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepository.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepository.java
index a345edb554f..cfe0544075e 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepository.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepository.java
@@ -63,6 +63,13 @@ public class StandardConnectorRepository implements
ConnectorRepository {
private static final Logger logger =
LoggerFactory.getLogger(StandardConnectorRepository.class);
private static final Duration SYNC_POLL_INTERVAL = Duration.ofSeconds(2);
+ /**
+ * Soft cardinality limit beyond which a WARN is emitted when a {@link
ConnectorConfigurationProvider}
+ * supplies logging attributes for a connector. Exceeding this threshold
signals a potential metric/MDC
+ * cardinality risk for downstream observability backends but does not
reject the attributes.
+ */
+ private static final int
CUSTOM_LOGGING_ATTRIBUTE_CARDINALITY_WARN_THRESHOLD = 5;
+
private final Map<String, ConnectorNode> connectors = new
ConcurrentHashMap<>();
private final FlowEngine lifecycleExecutor = new FlowEngine(8, "NiFi
Connector Lifecycle");
@@ -100,12 +107,14 @@ public class StandardConnectorRepository implements
ConnectorRepository {
@Override
public void addConnector(final ConnectorNode connector) {
+ applyProviderLoggingAttributes(connector);
syncFromProvider(connector);
connectors.put(connector.getIdentifier(), connector);
}
@Override
public void restoreConnector(final ConnectorNode connector) {
+ applyProviderLoggingAttributes(connector);
connectors.put(connector.getIdentifier(), connector);
logger.debug("Successfully restored {}", connector);
}
@@ -832,6 +841,39 @@ public class StandardConnectorRepository implements
ConnectorRepository {
return secretsManager;
}
+ /**
+ * Sources provider-supplied logging attributes for the given connector
and applies them to the node so that
+ * they are merged into the MDC of the connector's {@link
org.apache.nifi.logging.ComponentLog} and the loggers
+ * of every component running inside its managed flow (and surfaced on
connector metrics). Invoked when the
+ * connector node is added or restored. The framework-managed identity
keys are always present regardless of
+ * what the provider supplies.
+ */
+ private void applyProviderLoggingAttributes(final ConnectorNode connector)
{
+
connector.setCustomLoggingAttributes(resolveProviderLoggingAttributes(connector.getIdentifier()));
+ }
+
+ private Map<String, String> resolveProviderLoggingAttributes(final String
connectorId) {
+ if (configurationProvider == null) {
+ return Map.of();
+ }
+ try {
+ final Map<String, String> attributes =
configurationProvider.getLoggingAttributes(connectorId);
+ if (attributes == null || attributes.isEmpty()) {
+ return Map.of();
+ }
+ if (attributes.size() >
CUSTOM_LOGGING_ATTRIBUTE_CARDINALITY_WARN_THRESHOLD) {
+ logger.warn("ConnectorConfigurationProvider [{}] supplied {}
logging attributes for connector [{}] which exceeds the soft threshold of {}; "
+ + "high-cardinality attributes can degrade MDC
logging and OpenTelemetry metric backends",
+ configurationProvider.getClass().getSimpleName(),
attributes.size(), connectorId,
CUSTOM_LOGGING_ATTRIBUTE_CARDINALITY_WARN_THRESHOLD);
+ }
+ return attributes;
+ } catch (final Exception e) {
+ logger.warn("ConnectorConfigurationProvider [{}] threw while
computing logging attributes for connector [{}]; using empty map: {}",
+ configurationProvider.getClass().getSimpleName(),
connectorId, e.getMessage(), e);
+ return Map.of();
+ }
+ }
+
@Override
public ConnectorStateTransition createStateTransition(final String type,
final String id) {
final String componentDescription = "StandardConnectorNode[id=" + id +
", type=" + type + "]";
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/ExtensionBuilder.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/ExtensionBuilder.java
index 45ef9980d4b..103a47bee19 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/ExtensionBuilder.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/ExtensionBuilder.java
@@ -521,10 +521,11 @@ public class ExtensionBuilder {
}
final String componentType = connector.getClass().getSimpleName();
- final ComponentLog componentLog = new SimpleProcessLogger(identifier,
connector, new StandardLoggingContext());
+ final StandardLoggingContext loggingContext = new
StandardLoggingContext();
+ final ComponentLog componentLog = new SimpleProcessLogger(identifier,
connector, loggingContext);
final ConnectorDetails connectorDetails = new
ConnectorDetails(connector, bundleCoordinate, componentLog);
- final ConnectorNode connectorNode = new StandardConnectorNode(
+ final StandardConnectorNode connectorNode = new StandardConnectorNode(
identifier,
flowController.getFlowManager(),
extensionManager,
@@ -538,6 +539,11 @@ public class ExtensionBuilder {
connectorValidationTrigger,
false
);
+ // Late-bind the logging context to the connector node so that MDC
attributes assembled by
+ // the node (framework keys + provider-supplied custom keys) flow
through the connector's
+ // own ComponentLog as well as the managed flow's nested components.
Provider-supplied custom
+ // attributes are sourced and applied by the ConnectorRepository when
the node is added/restored.
+ loggingContext.setComponent(connectorNode);
try {
initializeDefaultValues(connector,
connectorNode.getActiveFlowContext());
@@ -565,13 +571,14 @@ public class ExtensionBuilder {
final GhostConnector ghostConnector = new GhostConnector(identifier,
type, cause);
final String simpleClassName = type.contains(".") ?
StringUtils.substringAfterLast(type, ".") : type;
final String componentType = "(Missing) " + simpleClassName;
- final ComponentLog componentLog = new SimpleProcessLogger(identifier,
ghostConnector, new StandardLoggingContext());
+ final StandardLoggingContext loggingContext = new
StandardLoggingContext();
+ final ComponentLog componentLog = new SimpleProcessLogger(identifier,
ghostConnector, loggingContext);
final ConnectorDetails connectorDetails = new
ConnectorDetails(ghostConnector, bundleCoordinate, componentLog);
// If an instance class loader has been created for this connector,
remove it because it's no longer necessary.
extensionManager.removeInstanceClassLoader(identifier);
- final ConnectorNode connectorNode = new StandardConnectorNode(
+ final StandardConnectorNode connectorNode = new StandardConnectorNode(
identifier,
flowController.getFlowManager(),
extensionManager,
@@ -585,6 +592,7 @@ public class ExtensionBuilder {
connectorValidationTrigger,
true
);
+ loggingContext.setComponent(connectorNode);
// Initialize the ghost connector so that it can be properly
configured during flow synchronization
final FrameworkConnectorInitializationContext initContext =
createConnectorInitializationContext(managedProcessGroup, componentLog);
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
index 0804b1d5662..804de6818a4 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
@@ -810,11 +810,12 @@ public class FlowController implements
ReportingTaskProvider, FlowAnalysisRulePr
}
eventAccess = new StandardEventAccess(flowManager,
flowFileEventRepository, processScheduler, authorizer, provenanceRepository,
- auditService, analyticsEngine, flowFileRepository,
contentRepository);
+ auditService, analyticsEngine, flowFileRepository,
contentRepository, connectorRepository);
timerDrivenEngineRef.get().scheduleWithFixedDelay(() -> {
try {
- statusHistoryRepository.capture(getNodeStatusSnapshot(),
eventAccess.getControllerStatus(), getGarbageCollectionStatus(), new Date());
+ statusHistoryRepository.capture(getNodeStatusSnapshot(),
eventAccess.getControllerStatus(), eventAccess.getConnectorStatuses(),
+ getGarbageCollectionStatus(), new Date());
} catch (final Exception e) {
LOG.error("Failed to capture component stats for Stats
History", e);
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepository.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepository.java
index 2b84217fa4e..c03cc077f88 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepository.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepository.java
@@ -17,6 +17,7 @@
package org.apache.nifi.controller.status.history;
import org.apache.nifi.controller.status.ConnectionStatus;
+import org.apache.nifi.controller.status.ConnectorStatus;
import org.apache.nifi.controller.status.NodeStatus;
import org.apache.nifi.controller.status.ProcessGroupStatus;
import org.apache.nifi.controller.status.ProcessorStatus;
@@ -28,6 +29,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
@@ -110,6 +112,24 @@ public class VolatileComponentStatusRepository implements
StatusHistoryRepositor
lastCaptureTime = Math.max(lastCaptureTime, timestamp.getTime());
}
+ @Override
+ public synchronized void capture(final NodeStatus nodeStatus, final
ProcessGroupStatus rootGroupStatus, final Collection<ConnectorStatus>
connectorStatuses,
+ final List<GarbageCollectionStatus>
gcStatus, final Date timestamp) {
+ capture(nodeStatus, rootGroupStatus, gcStatus, timestamp);
+
+ // Connector-managed flows are siblings of the root group and
therefore are not reachable from
+ // rootGroupStatus. Capture each Connector's managed flow so that
components running inside it have
+ // status history just like components in the primary flow.
+ if (connectorStatuses != null) {
+ for (final ConnectorStatus connectorStatus : connectorStatuses) {
+ final ProcessGroupStatus connectorRootGroupStatus =
connectorStatus.getRootGroupStatus();
+ if (connectorRootGroupStatus != null) {
+ capture(connectorRootGroupStatus, timestamp);
+ }
+ }
+ }
+ }
+
private void capture(final ProcessGroupStatus groupStatus, final Date
timestamp) {
// Capture status for the ProcessGroup
final ComponentDetails groupDetails =
ComponentDetails.forProcessGroup(groupStatus);
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/reporting/StandardEventAccess.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/reporting/StandardEventAccess.java
index ec233778b65..3e12745328e 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/reporting/StandardEventAccess.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/reporting/StandardEventAccess.java
@@ -22,6 +22,10 @@ import org.apache.nifi.authorization.Authorizer;
import org.apache.nifi.authorization.RequestAction;
import org.apache.nifi.authorization.resource.Authorizable;
import org.apache.nifi.authorization.user.NiFiUser;
+import org.apache.nifi.components.connector.ConnectorNode;
+import org.apache.nifi.components.connector.ConnectorRepository;
+import org.apache.nifi.components.connector.ConnectorSyncMode;
+import org.apache.nifi.components.connector.FrameworkFlowContext;
import org.apache.nifi.controller.ProcessScheduler;
import org.apache.nifi.controller.ProcessorNode;
import org.apache.nifi.controller.flow.FlowManager;
@@ -31,6 +35,7 @@ import
org.apache.nifi.controller.repository.FlowFileEventRepository;
import org.apache.nifi.controller.repository.FlowFileRepository;
import org.apache.nifi.controller.repository.RepositoryStatusReport;
import org.apache.nifi.controller.repository.metrics.EmptyFlowFileEvent;
+import org.apache.nifi.controller.status.ConnectorStatus;
import org.apache.nifi.controller.status.ProcessGroupStatus;
import org.apache.nifi.controller.status.ProcessorStatus;
import org.apache.nifi.controller.status.analytics.StatusAnalyticsEngine;
@@ -41,6 +46,8 @@ import org.apache.nifi.provenance.ProvenanceRepository;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -55,10 +62,11 @@ public class StandardEventAccess extends
AbstractEventAccess implements UserAwar
private final AuditService auditService;
private final FlowFileRepository flowFileRepository;
private final ContentRepository contentRepository;
+ private final ConnectorRepository connectorRepository;
public StandardEventAccess(final FlowManager flowManager, final
FlowFileEventRepository flowFileEventRepository, final ProcessScheduler
processScheduler,
final Authorizer authorizer, final
ProvenanceRepository provenanceRepository, final AuditService auditService,
final StatusAnalyticsEngine statusAnalyticsEngine,
- final FlowFileRepository flowFileRepository,
final ContentRepository contentRepository) {
+ final FlowFileRepository flowFileRepository,
final ContentRepository contentRepository, final ConnectorRepository
connectorRepository) {
super(processScheduler, statusAnalyticsEngine, flowManager,
flowFileEventRepository);
this.flowFileEventRepository = flowFileEventRepository;
this.flowManager = flowManager;
@@ -67,6 +75,49 @@ public class StandardEventAccess extends AbstractEventAccess
implements UserAwar
this.auditService = auditService;
this.flowFileRepository = flowFileRepository;
this.contentRepository = contentRepository;
+ this.connectorRepository = connectorRepository;
+ }
+
+ /**
+ * Returns a {@link ConnectorStatus} for each registered connector, each
carrying the connector identity along with
+ * the {@link ProcessGroupStatus} of the connector's managed root process
group. Connector-managed flows live
+ * outside the controller's root process group, so they are invisible to
{@link #getControllerStatus()} and would
+ * otherwise be skipped by standard reporting tasks.
+ */
+ @Override
+ public Collection<ConnectorStatus> getConnectorStatuses() {
+ if (connectorRepository == null) {
+ return Collections.emptyList();
+ }
+
+ final List<ConnectorNode> connectors =
connectorRepository.getConnectors(ConnectorSyncMode.LOCAL_ONLY);
+ if (connectors == null || connectors.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ final RepositoryStatusReport statusReport =
generateRepositoryStatusReport();
+ final List<ConnectorStatus> statuses = new
ArrayList<>(connectors.size());
+ for (final ConnectorNode connector : connectors) {
+ final FrameworkFlowContext context =
connector.getActiveFlowContext();
+ if (context == null) {
+ continue;
+ }
+ final ProcessGroup managedGroup = context.getManagedProcessGroup();
+ if (managedGroup == null) {
+ continue;
+ }
+ final ProcessGroupStatus rootGroupStatus =
getGroupStatus(managedGroup, statusReport,
+ authorizable -> true, Integer.MAX_VALUE, 1, true);
+ if (rootGroupStatus == null) {
+ continue;
+ }
+ final ConnectorStatus connectorStatus = new ConnectorStatus();
+ connectorStatus.setId(connector.getIdentifier());
+ connectorStatus.setName(connector.getName());
+ connectorStatus.setRootGroupStatus(rootGroupStatus);
+ statuses.add(connectorStatus);
+ }
+ return statuses;
}
@Override
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorNode.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorNode.java
index 8f4ad44d0aa..bbd0497a257 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorNode.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorNode.java
@@ -829,6 +829,88 @@ public class TestStandardConnectorNode {
assertEquals(ConnectorState.STOPPED, connectorNode.getCurrentState());
}
+ @Test
+ public void testLoggingAttributesIncludeFrameworkKeys() throws Exception {
+ final SleepingConnector connector = new
SleepingConnector(Duration.ofMillis(1));
+ final StandardConnectorNode connectorNode =
createConnectorNode(connector);
+
+ final Map<String, String> attributes =
connectorNode.getLoggingAttributes();
+ assertEquals("test-connector-id", attributes.get("connectorId"));
+ assertEquals(connector.getClass().getCanonicalName(),
attributes.get("connectorComponent"));
+ assertEquals("org.apache.nifi",
attributes.get("connectorBundleGroup"));
+ assertEquals("test-standard-connector-node",
attributes.get("connectorBundleArtifact"));
+ assertEquals("1.0.0", attributes.get("connectorBundleVersion"));
+ assertEquals(connector.getClass().getSimpleName(),
attributes.get("connectorName"));
+ }
+
+ @Test
+ public void testSetNameRefreshesConnectorNameLoggingAttribute() throws
Exception {
+ final StandardConnectorNode connectorNode = createConnectorNode();
+ connectorNode.setName("Renamed Connector");
+
+ assertEquals("Renamed Connector",
connectorNode.getLoggingAttributes().get("connectorName"));
+ }
+
+ @Test
+ public void testSetCustomLoggingAttributesMergesWithFrameworkKeys() throws
Exception {
+ final StandardConnectorNode connectorNode = createConnectorNode();
+ connectorNode.setCustomLoggingAttributes(Map.of(
+ "customA", "valueA",
+ "customB", "valueB"
+ ));
+
+ final Map<String, String> attributes =
connectorNode.getLoggingAttributes();
+ assertEquals("valueA", attributes.get("customA"));
+ assertEquals("valueB", attributes.get("customB"));
+ assertEquals("test-connector-id", attributes.get("connectorId"));
+ }
+
+ @Test
+ public void testSetCustomLoggingAttributesFiltersReservedKeys() throws
Exception {
+ final StandardConnectorNode connectorNode = createConnectorNode();
+ connectorNode.setCustomLoggingAttributes(Map.of(
+ "connectorId", "attempted-override",
+ "connectorBundleVersion", "9.9.9",
+ "customA", "valueA"
+ ));
+
+ final Map<String, String> attributes =
connectorNode.getLoggingAttributes();
+ // Reserved keys are owned by the framework: connectorId stays as the
framework value.
+ assertEquals("test-connector-id", attributes.get("connectorId"));
+ assertEquals("1.0.0", attributes.get("connectorBundleVersion"));
+ // Non-reserved custom keys are preserved.
+ assertEquals("valueA", attributes.get("customA"));
+ }
+
+ @Test
+ public void testSetCustomLoggingAttributesNullClears() throws Exception {
+ final StandardConnectorNode connectorNode = createConnectorNode();
+ connectorNode.setCustomLoggingAttributes(Map.of("customA", "valueA"));
+
assertTrue(connectorNode.getLoggingAttributes().containsKey("customA"));
+
+ connectorNode.setCustomLoggingAttributes(null);
+
assertFalse(connectorNode.getLoggingAttributes().containsKey("customA"));
+ // Framework keys remain.
+ assertEquals("test-connector-id",
connectorNode.getLoggingAttributes().get("connectorId"));
+ }
+
+ @Test
+ public void testReservedKeysExposeFrameworkAttributeNames() {
+ final Set<String> reserved =
ConnectorLoggingAttribute.getReservedKeys();
+ assertTrue(reserved.contains("connectorId"));
+ assertTrue(reserved.contains("connectorName"));
+ assertTrue(reserved.contains("connectorComponent"));
+ assertTrue(reserved.contains("connectorBundleGroup"));
+ assertTrue(reserved.contains("connectorBundleArtifact"));
+ assertTrue(reserved.contains("connectorBundleVersion"));
+ }
+
+ @Test
+ public void testGetProcessGroupReturnsManagedFlowRoot() throws Exception {
+ final StandardConnectorNode connectorNode = createConnectorNode();
+ assertEquals(managedProcessGroup, connectorNode.getProcessGroup());
+ }
+
private StandardConnectorNode createConnectorNode() throws
FlowUpdateException {
final SleepingConnector sleepingConnector = new
SleepingConnector(Duration.ofMillis(1));
return createConnectorNode(sleepingConnector);
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorRepository.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorRepository.java
index 4396b8a1ec9..68020bb8677 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorRepository.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorRepository.java
@@ -308,6 +308,10 @@ public class TestStandardConnectorRepository {
final ConnectorNode connector =
createSimpleConnectorNode("connector-1", "Local Name");
repository.restoreConnector(connector);
+ // Restoring the connector sources provider-supplied logging
attributes; reset so that the assertions
+ // below verify only that the LOCAL_ONLY get path does not consult the
provider.
+ reset(provider);
+
final ConnectorNode result = repository.getConnector("connector-1",
ConnectorSyncMode.LOCAL_ONLY);
assertNotNull(result);
@@ -325,6 +329,10 @@ public class TestStandardConnectorRepository {
repository.restoreConnector(connector1);
repository.restoreConnector(connector2);
+ // Restoring the connectors sources provider-supplied logging
attributes; reset so that the assertions
+ // below verify only that the LOCAL_ONLY get path does not consult the
provider.
+ reset(provider);
+
final List<ConnectorNode> results =
repository.getConnectors(ConnectorSyncMode.LOCAL_ONLY);
assertEquals(2, results.size());
@@ -1586,6 +1594,100 @@ public class TestStandardConnectorRepository {
return node;
}
+ // --- provider-sourced logging attribute application (add/restore) tests
---
+
+ @Test
+ public void testRestoreConnectorAppliesEmptyAttributesWhenNoProvider() {
+ final StandardConnectorRepository repository = new
StandardConnectorRepository();
+ final ConnectorRepositoryInitializationContext initContext =
mock(ConnectorRepositoryInitializationContext.class);
+
when(initContext.getExtensionManager()).thenReturn(mock(ExtensionManager.class));
+ when(initContext.getConnectorConfigurationProvider()).thenReturn(null);
+ repository.initialize(initContext);
+
+ final ConnectorNode connector =
createSimpleConnectorNode("connector-1", "Connector 1");
+ repository.restoreConnector(connector);
+
+ verify(connector).setCustomLoggingAttributes(Map.of());
+ }
+
+ @Test
+ public void testRestoreConnectorAppliesProviderLoggingAttributes() {
+ final ConnectorConfigurationProvider provider =
mock(ConnectorConfigurationProvider.class);
+ final Map<String, String> expected = Map.of("connectorDefinitionId",
"snowflake.runtime.postgres-cdc");
+
when(provider.getLoggingAttributes("connector-1")).thenReturn(expected);
+
+ final StandardConnectorRepository repository =
createRepositoryWithProvider(provider);
+
+ final ConnectorNode connector =
createSimpleConnectorNode("connector-1", "Connector 1");
+ repository.restoreConnector(connector);
+
+ verify(provider).getLoggingAttributes("connector-1");
+ verify(connector).setCustomLoggingAttributes(expected);
+ }
+
+ @Test
+ public void testAddConnectorAppliesProviderLoggingAttributes() {
+ final ConnectorConfigurationProvider provider =
mock(ConnectorConfigurationProvider.class);
+ final Map<String, String> expected = Map.of("connectorDefinitionId",
"snowflake.runtime.postgres-cdc");
+
when(provider.getLoggingAttributes("connector-1")).thenReturn(expected);
+
+ final StandardConnectorRepository repository =
createRepositoryWithProvider(provider);
+
+ final ConnectorNode connector =
createSimpleConnectorNode("connector-1", "Connector 1");
+ repository.addConnector(connector);
+
+ verify(provider).getLoggingAttributes("connector-1");
+ verify(connector).setCustomLoggingAttributes(expected);
+ }
+
+ @Test
+ public void
testRestoreConnectorAppliesEmptyAttributesWhenProviderReturnsNull() {
+ final ConnectorConfigurationProvider provider =
mock(ConnectorConfigurationProvider.class);
+ when(provider.getLoggingAttributes(anyString())).thenReturn(null);
+
+ final StandardConnectorRepository repository =
createRepositoryWithProvider(provider);
+
+ final ConnectorNode connector =
createSimpleConnectorNode("connector-1", "Connector 1");
+ repository.restoreConnector(connector);
+
+ verify(connector).setCustomLoggingAttributes(Map.of());
+ }
+
+ @Test
+ public void testRestoreConnectorAppliesEmptyAttributesWhenProviderThrows()
{
+ final ConnectorConfigurationProvider provider =
mock(ConnectorConfigurationProvider.class);
+ when(provider.getLoggingAttributes(anyString())).thenThrow(new
RuntimeException("provider boom"));
+
+ final StandardConnectorRepository repository =
createRepositoryWithProvider(provider);
+
+ final ConnectorNode connector =
createSimpleConnectorNode("connector-1", "Connector 1");
+ repository.restoreConnector(connector);
+
+ verify(connector).setCustomLoggingAttributes(Map.of());
+ }
+
+ @Test
+ public void
testRestoreConnectorAppliesAllAttributesWhenAboveCardinalityThreshold() {
+ // More than the soft cardinality threshold (5): the framework logs a
WARN but must not drop attributes.
+ final Map<String, String> expected = Map.of(
+ "attr1", "v1",
+ "attr2", "v2",
+ "attr3", "v3",
+ "attr4", "v4",
+ "attr5", "v5",
+ "attr6", "v6"
+ );
+ final ConnectorConfigurationProvider provider =
mock(ConnectorConfigurationProvider.class);
+
when(provider.getLoggingAttributes("connector-1")).thenReturn(expected);
+
+ final StandardConnectorRepository repository =
createRepositoryWithProvider(provider);
+
+ final ConnectorNode connector =
createSimpleConnectorNode("connector-1", "Connector 1");
+ repository.restoreConnector(connector);
+
+ verify(connector).setCustomLoggingAttributes(expected);
+ }
+
// --- Helper Methods ---
private StandardConnectorRepository
createRepositoryWithProviderAndAssetManager(
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/status/history/AbstractStatusHistoryRepositoryTest.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/status/history/AbstractStatusHistoryRepositoryTest.java
index 62f865721c9..2b032e62b89 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/status/history/AbstractStatusHistoryRepositoryTest.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/status/history/AbstractStatusHistoryRepositoryTest.java
@@ -17,6 +17,7 @@
package org.apache.nifi.controller.status.history;
import org.apache.nifi.controller.status.ConnectionStatus;
+import org.apache.nifi.controller.status.ConnectorStatus;
import org.apache.nifi.controller.status.NodeStatus;
import org.apache.nifi.controller.status.ProcessGroupStatus;
import org.apache.nifi.controller.status.ProcessorStatus;
@@ -41,6 +42,9 @@ public abstract class AbstractStatusHistoryRepositoryTest {
protected static final String PROCESSOR_WITH_COUNTER_ID =
"f09c5bb8-7b67-4d3b-81c5-5373e1c03ed1";
protected static final String CONNECTION_ID =
"d5452f70-a0d9-44c5-aeda-c2026482e4ee";
protected static final String REMOTE_PROCESS_GROUP_ID =
"f5f4fe2a-0209-4ba7-8f15-f33df942cde5";
+ protected static final String CONNECTOR_ID =
"a1111111-1111-1111-1111-111111111111";
+ protected static final String CONNECTOR_ROOT_GROUP_ID =
"a2222222-2222-2222-2222-222222222222";
+ protected static final String CONNECTOR_PROCESSOR_ID =
"a3333333-3333-3333-3333-333333333333";
protected ProcessGroupStatus givenSimpleRootProcessGroupStatus() {
final ProcessGroupStatus status = new ProcessGroupStatus();
@@ -74,6 +78,23 @@ public abstract class AbstractStatusHistoryRepositoryTest {
return status;
}
+ protected ConnectorStatus givenConnectorStatus() {
+ final ProcessorStatus connectorProcessor = givenProcessorStatus();
+ connectorProcessor.setId(CONNECTOR_PROCESSOR_ID);
+ connectorProcessor.setName("ConnectorProcessor");
+
+ final ProcessGroupStatus connectorRootGroup =
givenSimpleRootProcessGroupStatus();
+ connectorRootGroup.setId(CONNECTOR_ROOT_GROUP_ID);
+ connectorRootGroup.setName("Connector Root");
+
connectorRootGroup.setProcessorStatus(Collections.singleton(connectorProcessor));
+
+ final ConnectorStatus connectorStatus = new ConnectorStatus();
+ connectorStatus.setId(CONNECTOR_ID);
+ connectorStatus.setName("Test Connector");
+ connectorStatus.setRootGroupStatus(connectorRootGroup);
+ return connectorStatus;
+ }
+
protected ProcessGroupStatus givenChildProcessGroupStatus() {
final ProcessGroupStatus status = new ProcessGroupStatus();
status.setId(CHILD_GROUP_ID);
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepositoryForNodeTest.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepositoryForNodeTest.java
index 30681f4305c..0b5f363fea4 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepositoryForNodeTest.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepositoryForNodeTest.java
@@ -28,6 +28,7 @@ import java.util.List;
import java.util.Set;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.fail;
public class VolatileComponentStatusRepositoryForNodeTest extends
AbstractStatusHistoryRepositoryTest {
@@ -129,6 +130,34 @@ public class VolatileComponentStatusRepositoryForNodeTest
extends AbstractStatus
assertEquals(3L, getMetricAtOrdinal(snapshot2, g1CountDiffOrdinal));
}
+ @Test
+ public void testConnectorManagedFlowStatusHistory() {
+ final NiFiProperties niFiProperties =
Mockito.mock(NiFiProperties.class);
+
Mockito.when(niFiProperties.getIntegerProperty(VolatileComponentStatusRepository.NUM_DATA_POINTS_PROPERTY,
VolatileComponentStatusRepository.DEFAULT_NUM_DATA_POINTS)).thenReturn(10);
+ final VolatileComponentStatusRepository testSubject = new
VolatileComponentStatusRepository(niFiProperties);
+
+ final Date capturedAt = new Date();
+ testSubject.capture(givenNodeStatus(0), givenRootProcessGroupStatus(),
List.of(givenConnectorStatus()), givenGarbageCollectionStatuses(capturedAt),
capturedAt);
+
+ // The Connector's managed root group is captured even though it is a
sibling of (not reachable from) the root group
+ final StatusHistory connectorRootGroupStatus =
testSubject.getProcessGroupStatusHistory(CONNECTOR_ROOT_GROUP_ID, new Date(0),
new Date(), Integer.MAX_VALUE);
+ assertFalse(connectorRootGroupStatus.getStatusSnapshots().isEmpty());
+ assertEquals(CONNECTOR_ROOT_GROUP_ID,
connectorRootGroupStatus.getComponentDetails().get("Id"));
+ assertEquals("Connector Root",
connectorRootGroupStatus.getComponentDetails().get("Name"));
+
+ // A processor running inside the Connector-managed flow has status
history just like a primary-flow processor
+ final StatusHistory connectorProcessorStatus =
testSubject.getProcessorStatusHistory(CONNECTOR_PROCESSOR_ID, new Date(0), new
Date(), Integer.MAX_VALUE, false);
+ assertFalse(connectorProcessorStatus.getStatusSnapshots().isEmpty());
+ assertEquals(CONNECTOR_PROCESSOR_ID,
connectorProcessorStatus.getComponentDetails().get("Id"));
+ assertEquals("ConnectorProcessor",
connectorProcessorStatus.getComponentDetails().get("Name"));
+
+ // The primary root flow is still captured alongside the connector flow
+ final StatusHistory rootGroupStatus =
testSubject.getProcessGroupStatusHistory(ROOT_GROUP_ID, new Date(0), new
Date(), Integer.MAX_VALUE);
+ assertFalse(rootGroupStatus.getStatusSnapshots().isEmpty());
+ assertEquals(ROOT_GROUP_ID,
rootGroupStatus.getComponentDetails().get("Id"));
+ assertEquals("Root",
rootGroupStatus.getComponentDetails().get("Name"));
+ }
+
private static long getMetricAtOrdinal(final StatusSnapshot snapshot,
final long ordinal) {
final Set<MetricDescriptor<?>> metricDescriptors =
snapshot.getMetricDescriptors();
