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 a5bc2d3cea9 NIFI-16139 Added Source and Destination to Connection
Status Events (#11468)
a5bc2d3cea9 is described below
commit a5bc2d3cea95e834511365534ffac09f41d3c345
Author: David Handermann <[email protected]>
AuthorDate: Fri Jul 24 08:59:08 2026 -0500
NIFI-16139 Added Source and Destination to Connection Status Events (#11468)
---
.../controller/metrics/ConnectionStatusEvent.java | 14 ++++++++
.../repository/StandardProcessSession.java | 30 +++++++++++++++-
.../metrics/ConnectionStatusEventBuilder.java | 20 +++++++++--
.../metrics/StandardConnectionStatusEvent.java | 12 +++++++
.../repository/StandardProcessSessionTest.java | 42 +++++++++++++++++++---
5 files changed, 110 insertions(+), 8 deletions(-)
diff --git
a/nifi-framework-api/src/main/java/org/apache/nifi/controller/metrics/ConnectionStatusEvent.java
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/metrics/ConnectionStatusEvent.java
index d372489d59f..b1ed777b0e8 100644
---
a/nifi-framework-api/src/main/java/org/apache/nifi/controller/metrics/ConnectionStatusEvent.java
+++
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/metrics/ConnectionStatusEvent.java
@@ -30,6 +30,20 @@ public interface ConnectionStatusEvent {
*/
ComponentMetricContext getComponentMetricContext();
+ /**
+ * Get Component Metric Context describing the Source Component for the
Connection
+ *
+ * @return Source Component Metric Context
+ */
+ ComponentMetricContext getSourceComponentMetricContext();
+
+ /**
+ * Get Component Metric Context describing the Destination Component for
the Connection
+ *
+ * @return Destination Component Metric Context
+ */
+ ComponentMetricContext getDestinationComponentMetricContext();
+
/**
* Get configured Back Pressure Bytes Threshold
*
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
index 28e200d25c5..baa398ec827 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
@@ -52,6 +52,7 @@ import org.apache.nifi.controller.status.FlowFileAvailability;
import org.apache.nifi.controller.status.LoadBalanceStatus;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.processor.DataUnit;
import org.apache.nifi.processor.FlowFileFilter;
import org.apache.nifi.processor.ProcessSession;
@@ -824,7 +825,19 @@ public class StandardProcessSession implements
ProcessSession, ProvenanceEventEn
private void recordConnectionStatusEvents(final Checkpoint checkpoint) {
// Check enabled status to avoid building objects and calling methods
when not used
if (context.isRecordConnectionStatusEventEnabled()) {
+ final Map<String, ComponentMetricContext>
connectableMetricContexts = new HashMap<>();
+
for (final Connection connection :
checkpoint.processedConnections.values()) {
+ final Connectable source = connection.getSource();
+ final ComponentMetricContext sourceContext =
connectableMetricContexts.computeIfAbsent(source.getIdentifier(),
+ id -> getComponentMetricContext(source)
+ );
+
+ final Connectable destination = connection.getDestination();
+ final ComponentMetricContext destinationContext =
connectableMetricContexts.computeIfAbsent(destination.getIdentifier(),
+ id -> getComponentMetricContext(destination)
+ );
+
final ComponentMetricContext connectionMetricContext =
checkpoint.connectionMetricContexts.get(connection.getIdentifier());
final FlowFileQueue flowFileQueue =
connection.getFlowFileQueue();
final QueueSize queueSize = flowFileQueue.size();
@@ -832,7 +845,11 @@ public class StandardProcessSession implements
ProcessSession, ProvenanceEventEn
final LoadBalanceStatus loadBalanceStatus =
getLoadBalanceStatus(flowFileQueue);
final FlowFileAvailability flowFileAvailability =
flowFileQueue.getFlowFileAvailability();
- final ConnectionStatusEvent connectionStatusEvent =
ConnectionStatusEventBuilder.forComponent(connectionMetricContext)
+ final ConnectionStatusEvent connectionStatusEvent =
ConnectionStatusEventBuilder.forComponent(
+ connectionMetricContext,
+ sourceContext,
+ destinationContext
+ )
.backPressureBytesThreshold(backPressureBytesThreshold)
.backPressureObjectThreshold(flowFileQueue.getBackPressureObjectThreshold())
.queuedBytes(queueSize.getByteCount())
@@ -845,6 +862,17 @@ public class StandardProcessSession implements
ProcessSession, ProvenanceEventEn
}
}
+ private ComponentMetricContext getComponentMetricContext(final Connectable
connectable) {
+ final ProcessGroup processGroup = connectable.getProcessGroup();
+ final Map<String, String> attributes = processGroup == null ? Map.of()
: processGroup.getLoggingAttributes();
+ return new ComponentMetricContext(
+ connectable.getIdentifier(),
+ connectable.getName(),
+ connectable.getComponentType(),
+ attributes
+ );
+ }
+
private LoadBalanceStatus getLoadBalanceStatus(final FlowFileQueue
flowFileQueue) {
final LoadBalanceStatus loadBalanceStatus;
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/metrics/ConnectionStatusEventBuilder.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/metrics/ConnectionStatusEventBuilder.java
index 8c48a5e8e1a..515f1de4c8c 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/metrics/ConnectionStatusEventBuilder.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/metrics/ConnectionStatusEventBuilder.java
@@ -29,6 +29,8 @@ import java.util.Objects;
public class ConnectionStatusEventBuilder {
private final ComponentMetricContext componentMetricContext;
+ private final ComponentMetricContext sourceComponentMetricContext;
+ private final ComponentMetricContext destinationMetricContext;
private long backPressureBytesThreshold;
private long backPressureObjectThreshold;
@@ -37,12 +39,22 @@ public class ConnectionStatusEventBuilder {
private LoadBalanceStatus loadBalanceStatus =
LoadBalanceStatus.LOAD_BALANCE_NOT_CONFIGURED;
private FlowFileAvailability flowFileAvailability =
FlowFileAvailability.ACTIVE_QUEUE_EMPTY;
- private ConnectionStatusEventBuilder(final ComponentMetricContext
componentMetricContext) {
+ private ConnectionStatusEventBuilder(
+ final ComponentMetricContext componentMetricContext,
+ final ComponentMetricContext sourceComponentMetricContext,
+ final ComponentMetricContext destinationMetricContext
+ ) {
this.componentMetricContext =
Objects.requireNonNull(componentMetricContext, "Component Metric Context
required");
+ this.sourceComponentMetricContext =
Objects.requireNonNull(sourceComponentMetricContext, "Source Component Metric
Context required");
+ this.destinationMetricContext =
Objects.requireNonNull(destinationMetricContext, "Destination Component Metric
Context required");
}
- public static ConnectionStatusEventBuilder forComponent(final
ComponentMetricContext componentMetricContext) {
- return new ConnectionStatusEventBuilder(componentMetricContext);
+ public static ConnectionStatusEventBuilder forComponent(
+ final ComponentMetricContext componentMetricContext,
+ final ComponentMetricContext sourceComponentMetricContext,
+ final ComponentMetricContext destinationMetricContext
+ ) {
+ return new ConnectionStatusEventBuilder(componentMetricContext,
sourceComponentMetricContext, destinationMetricContext);
}
public ConnectionStatusEventBuilder backPressureBytesThreshold(final long
backPressureBytesThreshold) {
@@ -78,6 +90,8 @@ public class ConnectionStatusEventBuilder {
public ConnectionStatusEvent build() {
return new StandardConnectionStatusEvent(
componentMetricContext,
+ sourceComponentMetricContext,
+ destinationMetricContext,
backPressureBytesThreshold,
backPressureObjectThreshold,
queuedBytes,
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/metrics/StandardConnectionStatusEvent.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/metrics/StandardConnectionStatusEvent.java
index 4f4b32a20a8..008bc295ddb 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/metrics/StandardConnectionStatusEvent.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/metrics/StandardConnectionStatusEvent.java
@@ -26,6 +26,8 @@ import org.apache.nifi.controller.status.LoadBalanceStatus;
*/
record StandardConnectionStatusEvent(
ComponentMetricContext componentMetricContext,
+ ComponentMetricContext sourceComponentMetricContext,
+ ComponentMetricContext destinationComponentMetricContext,
long backPressureBytesThreshold,
long backPressureObjectThreshold,
long queuedBytes,
@@ -39,6 +41,16 @@ record StandardConnectionStatusEvent(
return componentMetricContext;
}
+ @Override
+ public ComponentMetricContext getSourceComponentMetricContext() {
+ return sourceComponentMetricContext;
+ }
+
+ @Override
+ public ComponentMetricContext getDestinationComponentMetricContext() {
+ return destinationComponentMetricContext;
+ }
+
@Override
public long getBackPressureBytesThreshold() {
return backPressureBytesThreshold;
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/controller/repository/StandardProcessSessionTest.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/controller/repository/StandardProcessSessionTest.java
index 79ce3f98a0f..1516507e894 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/controller/repository/StandardProcessSessionTest.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/controller/repository/StandardProcessSessionTest.java
@@ -94,6 +94,11 @@ class StandardProcessSessionTest {
private static final String BACK_PRESSURE_DATA_SIZE_THRESHOLD = "1 MB";
private static final long BACK_PRESSURE_BYTES_THRESHOLD = 1048576;
+ private static final String SOURCE_CONNECTABLE_ID = "source-id";
+ private static final String SOURCE_CONNECTABLE_NAME = "source-name";
+ private static final String DESTINATION_CONNECTABLE_ID = "destination-id";
+ private static final String DESTINATION_CONNECTABLE_NAME =
"destination-name";
+
@Mock
RepositoryContext repositoryContext;
@@ -181,7 +186,7 @@ class StandardProcessSessionTest {
when(repositoryContext.getContentRepository()).thenReturn(contentRepository);
when(repositoryContext.isRecordConnectionStatusEventEnabled()).thenReturn(true);
- final Connection connection = mock(Connection.class);
+ final Connection connection = getConnection();
when(repositoryContext.getPollableConnections()).thenReturn(List.of(connection));
final FlowFileRecord flowFileRecord = mock(FlowFileRecord.class);
when(connection.poll(anySet())).thenReturn(flowFileRecord);
@@ -192,7 +197,7 @@ class StandardProcessSessionTest {
final FlowFile flowFile = session.get();
assertNotNull(flowFile);
- final Connection outputConnection = mock(Connection.class);
+ final Connection outputConnection = getConnection();
when(outputConnection.getIdentifier()).thenReturn(OUTPUT_CONNECTION_ID);
final FlowFileQueue outputFlowFileQueue = mock(FlowFileQueue.class);
when(outputFlowFileQueue.getBackPressureDataSizeThreshold()).thenReturn(BACK_PRESSURE_DATA_SIZE_THRESHOLD);
@@ -227,10 +232,12 @@ class StandardProcessSessionTest {
assertEquals(objectCount, firstConnectionStatusEvent.getQueuedCount());
assertEquals(byteCount, firstConnectionStatusEvent.getQueuedBytes());
assertEquals(LoadBalanceStatus.LOAD_BALANCE_INACTIVE,
firstConnectionStatusEvent.getLoadBalanceStatus());
+ assertSourceDestinationFound(firstConnectionStatusEvent);
final ConnectionStatusEvent secondConnectionStatusEvent =
events.getLast();
final ComponentMetricContext secondComponentMetricContext =
secondConnectionStatusEvent.getComponentMetricContext();
assertEquals(OUTPUT_CONNECTION_ID, secondComponentMetricContext.id());
+ assertSourceDestinationFound(secondConnectionStatusEvent);
}
@Test
@@ -239,7 +246,7 @@ class StandardProcessSessionTest {
when(repositoryContext.getContentRepository()).thenReturn(contentRepository);
when(repositoryContext.isRecordConnectionStatusEventEnabled()).thenReturn(true);
- final Connection connection = mock(Connection.class);
+ final Connection connection = getConnection();
when(repositoryContext.getPollableConnections()).thenReturn(List.of(connection));
final FlowFileRecord flowFileRecord = mock(FlowFileRecord.class);
when(connection.poll(anySet())).thenReturn(flowFileRecord);
@@ -272,7 +279,7 @@ class StandardProcessSessionTest {
when(repositoryContext.getContentRepository()).thenReturn(contentRepository);
when(repositoryContext.isRecordConnectionStatusEventEnabled()).thenReturn(true);
- final Connection connection = mock(Connection.class);
+ final Connection connection = getConnection();
when(repositoryContext.getPollableConnections()).thenReturn(List.of(connection));
final FlowFileRecord flowFileRecord = mock(FlowFileRecord.class);
when(connection.poll(anySet())).thenReturn(flowFileRecord);
@@ -308,6 +315,7 @@ class StandardProcessSessionTest {
assertEquals(byteCount, connectionStatusEvent.getQueuedBytes());
assertEquals(LoadBalanceStatus.LOAD_BALANCE_NOT_CONFIGURED,
connectionStatusEvent.getLoadBalanceStatus());
assertEquals(FlowFileAvailability.FLOWFILE_AVAILABLE,
connectionStatusEvent.getFlowFileAvailability());
+ assertSourceDestinationFound(connectionStatusEvent);
}
@Test
@@ -408,6 +416,16 @@ class StandardProcessSessionTest {
assertEquals(bytesWritten, processSessionEvent.getBytesWritten(),
"Process Session Bytes written not matched");
}
+ private void assertSourceDestinationFound(final ConnectionStatusEvent
connectionStatusEvent) {
+ final ComponentMetricContext source =
connectionStatusEvent.getSourceComponentMetricContext();
+ assertEquals(SOURCE_CONNECTABLE_ID, source.id());
+ assertEquals(SOURCE_CONNECTABLE_NAME, source.name());
+
+ final ComponentMetricContext destination =
connectionStatusEvent.getDestinationComponentMetricContext();
+ assertEquals(DESTINATION_CONNECTABLE_ID, destination.id());
+ assertEquals(DESTINATION_CONNECTABLE_NAME, destination.name());
+ }
+
private void setRepositoryContext() {
when(repositoryContext.getProvenanceRepository()).thenReturn(provenanceRepository);
when(repositoryContext.getFlowFileRepository()).thenReturn(flowFileRepository);
@@ -420,4 +438,20 @@ class StandardProcessSessionTest {
destination.toFile().deleteOnExit();
return destination;
}
+
+ private Connection getConnection() {
+ final Connection connection = mock(Connection.class);
+
+ final Connectable source = mock(Connectable.class);
+ when(source.getIdentifier()).thenReturn(SOURCE_CONNECTABLE_ID);
+ when(source.getName()).thenReturn(SOURCE_CONNECTABLE_NAME);
+ when(connection.getSource()).thenReturn(source);
+
+ final Connectable destination = mock(Connectable.class);
+
when(destination.getIdentifier()).thenReturn(DESTINATION_CONNECTABLE_ID);
+ when(destination.getName()).thenReturn(DESTINATION_CONNECTABLE_NAME);
+ when(connection.getDestination()).thenReturn(destination);
+
+ return connection;
+ }
}
