This is an automated email from the ASF dual-hosted git repository.
exceptionfactory 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 acf723204ed NIFI-16519 Remove coupling between component scheduling
and ComponentMetricContext lifecycle (#11859)
acf723204ed is described below
commit acf723204ed2c36cdcd67383d1248487cf3e9fbc
Author: boguszj <[email protected]>
AuthorDate: Thu Nov 5 18:30:30 2026 +0100
NIFI-16519 Remove coupling between component scheduling and
ComponentMetricContext lifecycle (#11859)
Signed-off-by: David Handermann <[email protected]>
---
.../repository/AbstractRepositoryContext.java | 12 +-
.../controller/repository/RepositoryContext.java | 3 -
.../repository/StandardProcessSession.java | 26 ++---
.../java/org/apache/nifi/util/Connectables.java | 9 ++
.../repository/StandardProcessSessionTest.java | 30 +++++
.../nifi/controller/tasks/ConnectableTask.java | 4 +-
.../nifi/controller/tasks/StatelessFlowTask.java | 19 ++--
.../repository/StandardRepositoryContextTest.java | 126 +++++++++++++++++++++
.../nifi/controller/tasks/TestConnectableTask.java | 46 +++++++-
.../flow/StandardStatelessFlowCurrent.java | 10 +-
.../stateless/session/StatelessProcessSession.java | 5 +-
11 files changed, 237 insertions(+), 53 deletions(-)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/AbstractRepositoryContext.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/AbstractRepositoryContext.java
index 29c8d6817e1..03bbc60e47d 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/AbstractRepositoryContext.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/AbstractRepositoryContext.java
@@ -56,7 +56,6 @@ public abstract class AbstractRepositoryContext implements
RepositoryContext {
private final ProvenanceEventRepository provenanceRepo;
private final AtomicLong connectionIndex;
private final StateManager stateManager;
- private final ComponentMetricContext componentMetricContext;
private final ContentClaimCreationContext contentClaimCreationContext;
private final FlowFileUpdateContext flowFileUpdateContext;
@@ -84,11 +83,6 @@ public abstract class AbstractRepositoryContext implements
RepositoryContext {
this.connectionIndex = connectionIndex;
this.stateManager = stateManager;
- this.componentMetricContext = new ComponentMetricContext(
- connectable.getIdentifier(),
- connectable.getName(),
- connectable.getComponentType(),
- connectable.getProcessGroup().getLoggingAttributes());
this.componentNameCounterContext = connectable.getName() + " (" +
connectable.getIdentifier() + ")";
this.componentTypeCounterContext = "All " +
connectable.getComponentType() + "'s";
@@ -104,11 +98,6 @@ public abstract class AbstractRepositoryContext implements
RepositoryContext {
return connectable;
}
- @Override
- public ComponentMetricContext getComponentMetricContext() {
- return componentMetricContext;
- }
-
/**
*
* @param relationship relationship
@@ -177,6 +166,7 @@ public abstract class AbstractRepositoryContext implements
RepositoryContext {
counterRepo.adjustCounter(componentNameCounterContext, name, delta);
counterRepo.adjustCounter(componentTypeCounterContext, name, delta);
+ final ComponentMetricContext componentMetricContext =
Connectables.createComponentMetricContext(connectable);
final CounterRecord counterRecord = new CounterRecord(name, delta,
attributes, Instant.now(), componentMetricContext);
componentMetricReporter.recordCounter(counterRecord);
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/RepositoryContext.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/RepositoryContext.java
index 30cdc53caa5..8f5d50cf81c 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/RepositoryContext.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/RepositoryContext.java
@@ -20,7 +20,6 @@ package org.apache.nifi.controller.repository;
import org.apache.nifi.components.state.StateManager;
import org.apache.nifi.connectable.Connectable;
import org.apache.nifi.connectable.Connection;
-import org.apache.nifi.controller.metrics.ComponentMetricContext;
import org.apache.nifi.controller.metrics.ConnectionStatusEvent;
import org.apache.nifi.controller.metrics.GaugeRecord;
import org.apache.nifi.controller.metrics.ProcessSessionEvent;
@@ -40,8 +39,6 @@ import java.util.function.Predicate;
public interface RepositoryContext {
Connectable getConnectable();
- ComponentMetricContext getComponentMetricContext();
-
Collection<Connection> getConnections(Relationship relationship);
List<Connection> getPollableConnections();
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 1175b636d44..561c25d79b5 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,7 +52,6 @@ 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;
@@ -77,6 +76,7 @@ import org.apache.nifi.stream.io.ByteCountingOutputStream;
import org.apache.nifi.stream.io.LimitingInputStream;
import org.apache.nifi.stream.io.NonFlushableOutputStream;
import org.apache.nifi.stream.io.StreamUtils;
+import org.apache.nifi.util.Connectables;
import org.apache.nifi.util.FormatUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -795,7 +795,8 @@ public class StandardProcessSession implements
ProcessSession, ProvenanceEventEn
final Map<String, Long> counters =
combineCounters(checkpoint.countersOnCommit, checkpoint.immediateCounters);
- final ProcessSessionEvent flowFileEvent =
ProcessSessionEventBuilder.forComponent(context.getComponentMetricContext())
+ final ComponentMetricContext componentMetricContext =
Connectables.createComponentMetricContext(context.getConnectable());
+ final ProcessSessionEvent flowFileEvent =
ProcessSessionEventBuilder.forComponent(componentMetricContext)
.flowFilesIn(checkpoint.flowFilesIn)
.flowFilesOut(checkpoint.flowFilesOut)
.flowFilesRemoved(checkpoint.removedCount)
@@ -833,12 +834,12 @@ public class StandardProcessSession implements
ProcessSession, ProvenanceEventEn
for (final Connection connection :
checkpoint.processedConnections.values()) {
final Connectable source = connection.getSource();
final ComponentMetricContext sourceContext =
connectableMetricContexts.computeIfAbsent(source.getIdentifier(),
- id -> getComponentMetricContext(source)
+ id -> Connectables.createComponentMetricContext(source)
);
final Connectable destination = connection.getDestination();
final ComponentMetricContext destinationContext =
connectableMetricContexts.computeIfAbsent(destination.getIdentifier(),
- id -> getComponentMetricContext(destination)
+ id ->
Connectables.createComponentMetricContext(destination)
);
final ComponentMetricContext connectionMetricContext =
checkpoint.connectionMetricContexts.get(connection.getIdentifier());
@@ -865,17 +866,6 @@ 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;
@@ -1422,7 +1412,8 @@ public class StandardProcessSession implements
ProcessSession, ProvenanceEventEn
}
}
- final ProcessSessionEvent flowFileEvent =
ProcessSessionEventBuilder.forComponent(context.getComponentMetricContext())
+ final ComponentMetricContext componentMetricContext =
Connectables.createComponentMetricContext(context.getConnectable());
+ final ProcessSessionEvent flowFileEvent =
ProcessSessionEventBuilder.forComponent(componentMetricContext)
.bytesRead(bytesRead)
.bytesWritten(bytesWritten)
.counters(getCounterValues(immediateCounters))
@@ -2035,7 +2026,8 @@ public class StandardProcessSession implements
ProcessSession, ProvenanceEventEn
Objects.requireNonNull(commitTiming, "Commit Timing required");
final Instant recorded = Instant.now();
- final GaugeRecord gaugeRecord = new GaugeRecord(name, value,
Map.copyOf(attributes), recorded, context.getComponentMetricContext());
+ final ComponentMetricContext componentMetricContext =
Connectables.createComponentMetricContext(context.getConnectable());
+ final GaugeRecord gaugeRecord = new GaugeRecord(name, value,
Map.copyOf(attributes), recorded, componentMetricContext);
if (CommitTiming.NOW == commitTiming) {
context.recordGauge(gaugeRecord);
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/util/Connectables.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/util/Connectables.java
index 4a36be5a85c..6fe1bb33c0c 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/util/Connectables.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/util/Connectables.java
@@ -18,11 +18,14 @@ package org.apache.nifi.util;
import org.apache.nifi.connectable.Connectable;
import org.apache.nifi.connectable.Connection;
+import org.apache.nifi.controller.metrics.ComponentMetricContext;
import org.apache.nifi.controller.status.FlowFileAvailability;
+import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.processor.Relationship;
import java.util.Collection;
import java.util.List;
+import java.util.Map;
public class Connectables {
@@ -66,4 +69,10 @@ public class Connectables {
return false;
}
+
+ public static ComponentMetricContext createComponentMetricContext(final
Connectable connectable) {
+ final ProcessGroup processGroup = connectable.getProcessGroup();
+ final Map<String, String> groupAttributes = processGroup == null ?
Map.of() : processGroup.getLoggingAttributes();
+ return new ComponentMetricContext(connectable.getIdentifier(),
connectable.getName(), connectable.getComponentType(), groupAttributes);
+ }
}
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 ad04a8ecfd6..f4b4001795a 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
@@ -33,6 +33,7 @@ import
org.apache.nifi.controller.repository.metrics.PerformanceTracker;
import org.apache.nifi.controller.status.FlowFileAvailability;
import org.apache.nifi.controller.status.LoadBalanceStatus;
import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.metrics.CommitTiming;
@@ -99,6 +100,10 @@ class StandardProcessSessionTest {
private static final Map<String, String> METRIC_ATTRIBUTES =
Map.of("service.name", "Processing", "deployment.environment", "production");
+ private static final Map<String, String> INITIAL_LOGGING_ATTRIBUTES =
Map.of("registeredFlowIdentifierPath", "/root:1/connector:1.0.0");
+
+ private static final Map<String, String> UPDATED_LOGGING_ATTRIBUTES =
Map.of("registeredFlowIdentifierPath", "/root:1/connector:2.0.0");
+
private static final String INPUT_CONNECTION_ID = "input-connection-id";
private static final String OUTPUT_CONNECTION_ID = "output-connection-id";
private static final String BACK_PRESSURE_DATA_SIZE_THRESHOLD = "1 MB";
@@ -391,6 +396,31 @@ class StandardProcessSessionTest {
assertGaugeRecordsMatched();
}
+ @Test
+ void testCommitRecordsCurrentProcessGroupLoggingAttributes() {
+ setRepositoryContext();
+ final ProcessGroup processGroup = mock(ProcessGroup.class);
+ when(connectable.getProcessGroup()).thenReturn(processGroup);
+
when(processGroup.getLoggingAttributes()).thenReturn(INITIAL_LOGGING_ATTRIBUTES);
+
+ session.recordGauge(GAUGE_NAME, GAUGE_VALUE,
CommitTiming.SESSION_COMMITTED);
+ session.commit();
+
+
when(processGroup.getLoggingAttributes()).thenReturn(UPDATED_LOGGING_ATTRIBUTES);
+ session.recordGauge(GAUGE_NAME, GAUGE_VALUE,
CommitTiming.SESSION_COMMITTED);
+ session.commit();
+
+ verify(repositoryContext,
times(2)).recordGauge(gaugeRecordCaptor.capture());
+ final List<GaugeRecord> gaugeRecords =
gaugeRecordCaptor.getAllValues();
+ assertEquals(INITIAL_LOGGING_ATTRIBUTES,
gaugeRecords.getFirst().componentMetricContext().attributes());
+ assertEquals(UPDATED_LOGGING_ATTRIBUTES,
gaugeRecords.getLast().componentMetricContext().attributes());
+
+ verify(repositoryContext,
times(2)).recordProcessSessionEvent(processSessionEventCaptor.capture());
+ final List<ProcessSessionEvent> processSessionEvents =
processSessionEventCaptor.getAllValues();
+ assertEquals(INITIAL_LOGGING_ATTRIBUTES,
processSessionEvents.getFirst().getComponentMetricContext().attributes());
+ assertEquals(UPDATED_LOGGING_ATTRIBUTES,
processSessionEvents.getLast().getComponentMetricContext().attributes());
+ }
+
@Test
void testAdjustCounterNow() {
session.adjustCounter(COUNTER_NAME, COUNTER_DELTA, true);
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ConnectableTask.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ConnectableTask.java
index 7c94e8a6163..7143ec9c298 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ConnectableTask.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ConnectableTask.java
@@ -24,6 +24,7 @@ import org.apache.nifi.controller.FlowController;
import org.apache.nifi.controller.ProcessorNode;
import org.apache.nifi.controller.ScheduledState;
import org.apache.nifi.controller.lifecycle.TaskTerminationAwareStateManager;
+import org.apache.nifi.controller.metrics.ComponentMetricContext;
import org.apache.nifi.controller.metrics.ProcessSessionEvent;
import org.apache.nifi.controller.repository.ActiveProcessSessionFactory;
import org.apache.nifi.controller.repository.BatchingSessionFactory;
@@ -321,7 +322,8 @@ public class ConnectableTask {
}
private void updateEventRepo(final TrackedStats stats, final int
invocationCount) throws IOException {
- final ProcessSessionEvent flowFileEvent =
stats.end(repositoryContext.getComponentMetricContext())
+ final ComponentMetricContext componentMetricContext =
Connectables.createComponentMetricContext(connectable);
+ final ProcessSessionEvent flowFileEvent =
stats.end(componentMetricContext)
.invocations(invocationCount)
.build();
repositoryContext.getFlowFileEventRepository().updateRepository(flowFileEvent);
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/StatelessFlowTask.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/StatelessFlowTask.java
index deae10e4ff2..40161d1bad3 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/StatelessFlowTask.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/StatelessFlowTask.java
@@ -17,7 +17,6 @@
package org.apache.nifi.controller.tasks;
-import org.apache.nifi.connectable.Connectable;
import org.apache.nifi.connectable.Connection;
import org.apache.nifi.connectable.ConnectionUtils;
import org.apache.nifi.connectable.ConnectionUtils.FlowFileCloneResult;
@@ -25,7 +24,6 @@ import org.apache.nifi.connectable.Port;
import org.apache.nifi.controller.ProcessorNode;
import org.apache.nifi.controller.ScheduledState;
import org.apache.nifi.controller.Triggerable;
-import org.apache.nifi.controller.metrics.ComponentMetricContext;
import org.apache.nifi.controller.queue.FlowFileQueue;
import org.apache.nifi.controller.repository.ContentRepository;
import org.apache.nifi.controller.repository.FlowFileEventRepository;
@@ -57,6 +55,7 @@ import org.apache.nifi.stateless.flow.FlowFileSupplier;
import org.apache.nifi.stateless.flow.StatelessDataflow;
import org.apache.nifi.stateless.flow.TriggerResult;
import org.apache.nifi.stateless.repository.StatelessProvenanceRepository;
+import org.apache.nifi.util.Connectables;
import java.io.IOException;
import java.time.Duration;
@@ -632,7 +631,9 @@ public class StatelessFlowTask {
final Port inputPort = polledFlowFile.getInputPort();
final ProcessSessionEventBuilder cumulativeEvent =
eventsByComponentId.computeIfAbsent(
- inputPort.getIdentifier(), id ->
ProcessSessionEventBuilder.forComponent(createMetricContext(inputPort)));
+ inputPort.getIdentifier(),
+ id ->
ProcessSessionEventBuilder.forComponent(Connectables.createComponentMetricContext(inputPort))
+ );
cumulativeEvent.addFlowFilesIn(1)
.addFlowFilesOut(numOutputConnections)
.addContentSizeIn(bytes)
@@ -654,7 +655,9 @@ public class StatelessFlowTask {
final int outputConnectionCount = port.getConnections().size();
final ProcessSessionEventBuilder cumulativeEvent =
eventsByComponentId.computeIfAbsent(
- port.getIdentifier(), id ->
ProcessSessionEventBuilder.forComponent(createMetricContext(port)));
+ port.getIdentifier(),
+ id ->
ProcessSessionEventBuilder.forComponent(Connectables.createComponentMetricContext(port))
+ );
cumulativeEvent.addFlowFilesIn(flowFileCount)
.addFlowFilesOut(flowFileCount * outputConnectionCount)
.addContentSizeIn(byteCount)
@@ -671,14 +674,6 @@ public class StatelessFlowTask {
}
}
- private static ComponentMetricContext createMetricContext(final
Connectable connectable) {
- final ProcessGroup processGroup = connectable.getProcessGroup();
- final Map<String, String> groupAttributes = processGroup == null ?
Map.of() : processGroup.getLoggingAttributes();
- return new ComponentMetricContext(
- connectable.getIdentifier(), connectable.getName(),
connectable.getComponentType(), groupAttributes);
- }
-
-
/**
* Updates the Stateless Group Node's FlowFileActivity. The latest
activity time is obtained from the stateless flow's root group,
* which is updated by processors within the flow when their sessions
commit. Transfer counts are computed from provenance events
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/StandardRepositoryContextTest.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/StandardRepositoryContextTest.java
new file mode 100644
index 00000000000..fa36e51703f
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/StandardRepositoryContextTest.java
@@ -0,0 +1,126 @@
+/*
+ * 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.controller.repository;
+
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.connectable.Connectable;
+import org.apache.nifi.controller.metrics.ComponentMetricContext;
+import org.apache.nifi.controller.metrics.ComponentMetricReporter;
+import org.apache.nifi.controller.metrics.CounterRecord;
+import org.apache.nifi.groups.ProcessGroup;
+import org.apache.nifi.provenance.ProvenanceEventRepository;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class StandardRepositoryContextTest {
+ private static final String COMPONENT_ID = "component-id";
+
+ private static final String COMPONENT_NAME = "Component Name";
+
+ private static final String COMPONENT_TYPE = "ComponentType";
+
+ private static final String COUNTER_NAME = "records.ingested";
+
+ private static final long COUNTER_DELTA = 5;
+
+ private static final Map<String, String> COUNTER_ATTRIBUTES =
Map.of("table", "orders");
+
+ private static final Map<String, String> INITIAL_LOGGING_ATTRIBUTES =
Map.of("registeredFlowIdentifierPath", "/root:1/connector:1.0.0");
+
+ private static final Map<String, String> UPDATED_LOGGING_ATTRIBUTES =
Map.of("registeredFlowIdentifierPath", "/root:1/connector:2.0.0");
+
+ @Mock
+ private Connectable connectable;
+
+ @Mock
+ private ProcessGroup processGroup;
+
+ @Mock
+ private ContentRepository contentRepository;
+
+ @Mock
+ private FlowFileRepository flowFileRepository;
+
+ @Mock
+ private FlowFileEventRepository flowFileEventRepository;
+
+ @Mock
+ private CounterRepository counterRepository;
+
+ @Mock
+ private ComponentMetricReporter componentMetricReporter;
+
+ @Mock
+ private ProvenanceEventRepository provenanceEventRepository;
+
+ @Mock
+ private StateManager stateManager;
+
+ @Captor
+ private ArgumentCaptor<CounterRecord> counterRecordCaptor;
+
+ @Test
+ void testAdjustCounterRecordsCurrentProcessGroupLoggingAttributes() {
+ when(connectable.getIdentifier()).thenReturn(COMPONENT_ID);
+ when(connectable.getName()).thenReturn(COMPONENT_NAME);
+ when(connectable.getComponentType()).thenReturn(COMPONENT_TYPE);
+ when(connectable.getProcessGroup()).thenReturn(processGroup);
+
when(processGroup.getLoggingAttributes()).thenReturn(INITIAL_LOGGING_ATTRIBUTES);
+
+ final StandardRepositoryContext repositoryContext = new
StandardRepositoryContext(
+ connectable,
+ new AtomicLong(),
+ contentRepository,
+ flowFileRepository,
+ flowFileEventRepository,
+ counterRepository,
+ componentMetricReporter,
+ provenanceEventRepository,
+ stateManager
+ );
+ repositoryContext.adjustCounter(COUNTER_NAME, COUNTER_DELTA,
COUNTER_ATTRIBUTES);
+
+
when(processGroup.getLoggingAttributes()).thenReturn(UPDATED_LOGGING_ATTRIBUTES);
+ repositoryContext.adjustCounter(COUNTER_NAME, COUNTER_DELTA,
COUNTER_ATTRIBUTES);
+
+ verify(componentMetricReporter,
times(2)).recordCounter(counterRecordCaptor.capture());
+ final List<CounterRecord> counterRecords =
counterRecordCaptor.getAllValues();
+ assertCounterRecordEquals(INITIAL_LOGGING_ATTRIBUTES,
counterRecords.getFirst());
+ assertCounterRecordEquals(UPDATED_LOGGING_ATTRIBUTES,
counterRecords.getLast());
+ }
+
+ private void assertCounterRecordEquals(final Map<String, String>
expectedLoggingAttributes, final CounterRecord counterRecord) {
+ assertEquals(COUNTER_NAME, counterRecord.name());
+ assertEquals(COUNTER_DELTA, counterRecord.value());
+ assertEquals(COUNTER_ATTRIBUTES, counterRecord.attributes());
+ assertEquals(new ComponentMetricContext(COMPONENT_ID, COMPONENT_NAME,
COMPONENT_TYPE, expectedLoggingAttributes),
counterRecord.componentMetricContext());
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/tasks/TestConnectableTask.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/tasks/TestConnectableTask.java
index a41da4b0c9a..ee6b1d4930c 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/tasks/TestConnectableTask.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/tasks/TestConnectableTask.java
@@ -26,6 +26,7 @@ import org.apache.nifi.controller.FlowController;
import org.apache.nifi.controller.GarbageCollectionLog;
import org.apache.nifi.controller.ProcessorNode;
import org.apache.nifi.controller.ScheduledState;
+import org.apache.nifi.controller.metrics.ProcessSessionEvent;
import org.apache.nifi.controller.queue.FlowFileQueue;
import org.apache.nifi.controller.repository.FlowFileEventRepository;
import org.apache.nifi.controller.repository.RepositoryContext;
@@ -34,34 +35,50 @@ import org.apache.nifi.controller.scheduling.LifecycleState;
import org.apache.nifi.controller.scheduling.RepositoryContextFactory;
import org.apache.nifi.controller.scheduling.SchedulingAgent;
import org.apache.nifi.controller.status.FlowFileAvailability;
+import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.processor.Processor;
import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
public class TestConnectableTask {
+ private static final Map<String, String> INITIAL_LOGGING_ATTRIBUTES =
Map.of("registeredFlowIdentifierPath", "/root:1/connector:1.0.0");
+
+ private static final Map<String, String> UPDATED_LOGGING_ATTRIBUTES =
Map.of("registeredFlowIdentifierPath", "/root:1/connector:2.0.0");
+
private ConnectableTask createTask(final Connectable connectable) {
+ return createTask(connectable, mock(FlowFileEventRepository.class));
+ }
+
+ private ConnectableTask createTask(final Connectable connectable, final
FlowFileEventRepository flowFileEventRepository) {
final FlowController flowController = mock(FlowController.class);
when(flowController.getStateManagerProvider()).thenReturn(mock(StateManagerProvider.class));
when(flowController.getExtensionManager()).thenReturn(mock(ExtensionManager.class));
final RepositoryContext repoContext =
mock(StandardRepositoryContext.class);
-
when(repoContext.getFlowFileEventRepository()).thenReturn(mock(FlowFileEventRepository.class));
+
when(repoContext.getFlowFileEventRepository()).thenReturn(flowFileEventRepository);
when(flowController.getGarbageCollectionLog()).thenReturn(mock(GarbageCollectionLog.class));
@@ -102,6 +119,33 @@ public class TestConnectableTask {
assertFalse(taskMonitorHeld.get());
}
+ @Test
+ public void testInvokeRecordsCurrentProcessGroupLoggingAttributes() throws
IOException {
+ final ProcessGroup processGroup = mock(ProcessGroup.class);
+
when(processGroup.getLoggingAttributes()).thenReturn(INITIAL_LOGGING_ATTRIBUTES);
+
+ final ProcessorNode processorNode = mock(ProcessorNode.class);
+ when(processorNode.getIdentifier()).thenReturn("processor-id");
+
when(processorNode.getRunnableComponent()).thenReturn(mock(Processor.class));
+
when(processorNode.getRelationships()).thenReturn(Collections.emptySet());
+
when(processorNode.getIncomingConnections()).thenReturn(Collections.emptyList());
+
when(processorNode.getScheduledState()).thenReturn(ScheduledState.RUNNING);
+ when(processorNode.getProcessGroup()).thenReturn(processGroup);
+
+ final FlowFileEventRepository flowFileEventRepository =
mock(FlowFileEventRepository.class);
+ final ConnectableTask task = createTask(processorNode,
flowFileEventRepository);
+ task.invoke();
+
+
when(processGroup.getLoggingAttributes()).thenReturn(UPDATED_LOGGING_ATTRIBUTES);
+ task.invoke();
+
+ final ArgumentCaptor<ProcessSessionEvent> eventCaptor =
ArgumentCaptor.forClass(ProcessSessionEvent.class);
+ verify(flowFileEventRepository,
times(2)).updateRepository(eventCaptor.capture());
+ final List<ProcessSessionEvent> events = eventCaptor.getAllValues();
+ assertEquals(INITIAL_LOGGING_ATTRIBUTES,
events.getFirst().getComponentMetricContext().attributes());
+ assertEquals(UPDATED_LOGGING_ATTRIBUTES,
events.getLast().getComponentMetricContext().attributes());
+ }
+
@Test
public void testIsWorkToDo() {
final ProcessorNode procNode = mock(ProcessorNode.class);
diff --git
a/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/flow/StandardStatelessFlowCurrent.java
b/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/flow/StandardStatelessFlowCurrent.java
index 0c8a094917d..a2e10521877 100644
---
a/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/flow/StandardStatelessFlowCurrent.java
+++
b/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/flow/StandardStatelessFlowCurrent.java
@@ -40,12 +40,12 @@ import
org.apache.nifi.stateless.engine.ProcessContextFactory;
import org.apache.nifi.stateless.repository.RepositoryContextFactory;
import org.apache.nifi.stateless.session.AsynchronousCommitTracker;
import org.apache.nifi.stateless.session.StatelessProcessSessionFactory;
+import org.apache.nifi.util.Connectables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Collection;
-import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
@@ -313,12 +313,8 @@ public class StandardStatelessFlowCurrent implements
StatelessFlowCurrent {
private void registerProcessEvent(final Connectable connectable, final
TrackedStats trackedStats) {
try {
- final Map<String, String> groupAttributes =
connectable.getProcessGroup() == null
- ? Map.of()
- : connectable.getProcessGroup().getLoggingAttributes();
- final ComponentMetricContext metricContext = new
ComponentMetricContext(
- connectable.getIdentifier(), connectable.getName(),
connectable.getComponentType(), groupAttributes);
- final ProcessSessionEvent procEvent =
trackedStats.end(metricContext)
+ final ComponentMetricContext componentMetricContext =
Connectables.createComponentMetricContext(connectable);
+ final ProcessSessionEvent procEvent =
trackedStats.end(componentMetricContext)
.invocations(1)
.build();
repositoryContextFactory.getFlowFileEventRepository().updateRepository(procEvent);
diff --git
a/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/session/StatelessProcessSession.java
b/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/session/StatelessProcessSession.java
index 0434bdebda3..efdcf1f3637 100644
---
a/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/session/StatelessProcessSession.java
+++
b/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/session/StatelessProcessSession.java
@@ -22,6 +22,7 @@ import org.apache.nifi.connectable.Connectable;
import org.apache.nifi.connectable.ConnectableType;
import org.apache.nifi.connectable.Connection;
import org.apache.nifi.connectable.Port;
+import org.apache.nifi.controller.metrics.ComponentMetricContext;
import org.apache.nifi.controller.metrics.ProcessSessionEvent;
import org.apache.nifi.controller.repository.FlowFileRecord;
import org.apache.nifi.controller.repository.StandardProcessSession;
@@ -38,6 +39,7 @@ import
org.apache.nifi.stateless.flow.FailurePortEncounteredException;
import org.apache.nifi.stateless.flow.StandardStatelessFlow;
import org.apache.nifi.stateless.queue.DrainableFlowFileQueue;
import org.apache.nifi.stateless.repository.RepositoryContextFactory;
+import org.apache.nifi.util.Connectables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -347,7 +349,8 @@ public class StatelessProcessSession extends
StandardProcessSession {
private void registerProcessEvent(final Connectable connectable, final
long processingNanos) {
try {
- final ProcessSessionEvent procEvent =
ProcessSessionEventBuilder.forComponent(getRepositoryContext().getComponentMetricContext())
+ final ComponentMetricContext componentMetricContext =
Connectables.createComponentMetricContext(connectable);
+ final ProcessSessionEvent procEvent =
ProcessSessionEventBuilder.forComponent(componentMetricContext)
.invocations(1)
.processingNanoseconds(processingNanos)
.build();
