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 59b3217e5fa NIFI-16453 Added component metric assertions to
ConnectorTestRunner (#11831)
59b3217e5fa is described below
commit 59b3217e5faeb2b1904b60fbacd0b374c551702f
Author: Adela Jaworowska-Nowak <[email protected]>
AuthorDate: Mon Oct 26 10:58:51 2026 +0100
NIFI-16453 Added component metric assertions to ConnectorTestRunner (#11831)
Describe null and empty results, and the difference between a summed lookup
and an exact attribute match. State that a runner without component metrics
does not support them.
---
.../mock/connector/server/ConnectorTestRunner.java | 50 ++++++++++++
.../server/RecordingComponentMetricReporter.java | 93 ++++++++++++++++++++++
.../server/StandardConnectorMockServer.java | 25 +++++-
.../RecordingComponentMetricReporterTest.java | 69 ++++++++++++++++
.../mock/connectors/tests/MockProcessorIT.java | 36 +++++++++
.../connector/StandardConnectorTestRunner.java | 20 +++++
6 files changed, 291 insertions(+), 2 deletions(-)
diff --git
a/nifi-connector-mock-bundle/nifi-connector-mock-api/src/main/java/org/apache/nifi/mock/connector/server/ConnectorTestRunner.java
b/nifi-connector-mock-bundle/nifi-connector-mock-api/src/main/java/org/apache/nifi/mock/connector/server/ConnectorTestRunner.java
index 9678330f676..e6c9a464abe 100644
---
a/nifi-connector-mock-bundle/nifi-connector-mock-api/src/main/java/org/apache/nifi/mock/connector/server/ConnectorTestRunner.java
+++
b/nifi-connector-mock-bundle/nifi-connector-mock-api/src/main/java/org/apache/nifi/mock/connector/server/ConnectorTestRunner.java
@@ -194,6 +194,56 @@ public interface ConnectorTestRunner extends Closeable {
*/
List<ValidationResult> validate();
+ /**
+ * Returns the recorded counter value for the component and counter name.
+ * Values recorded with different attribute maps are summed.
+ *
+ * @param componentId the identifier of the component that recorded the
counter
+ * @param name the counter name
+ * @return the summed counter value, or {@code null} when nothing was
recorded
+ */
+ default Long getCounterValue(final String componentId, final String name) {
+ throw new UnsupportedOperationException("This runner does not support
component metrics");
+ }
+
+ /**
+ * Returns the recorded counter value for the component, counter name, and
exact attribute map.
+ * Unlike {@link #getCounterValue(String, String)}, attribute maps are not
summed together.
+ *
+ * @param componentId the identifier of the component that recorded the
counter
+ * @param name the counter name
+ * @param attributes the attribute map that must match the recorded counter
+ * @return the counter value, or {@code null} when nothing was recorded
for that attribute map
+ */
+ default Long getCounterValue(final String componentId, final String name,
final Map<String, String> attributes) {
+ throw new UnsupportedOperationException("This runner does not support
component metrics");
+ }
+
+ /**
+ * Returns the gauge samples recorded for the component and gauge name, in
the order they were recorded.
+ * Samples recorded with different attribute maps are included.
+ *
+ * @param componentId the identifier of the component that recorded the
gauge
+ * @param name the gauge name
+ * @return the recorded samples, or an empty list when nothing was recorded
+ */
+ default List<Double> getGaugeValues(final String componentId, final String
name) {
+ throw new UnsupportedOperationException("This runner does not support
component metrics");
+ }
+
+ /**
+ * Returns the gauge samples recorded for the component, gauge name, and
exact attribute map, in the order they were recorded.
+ * Unlike {@link #getGaugeValues(String, String)}, samples with a
different attribute map are excluded.
+ *
+ * @param componentId the identifier of the component that recorded the
gauge
+ * @param name the gauge name
+ * @param attributes the attribute map that must match the recorded gauge
+ * @return the recorded samples, or an empty list when nothing was
recorded for that attribute map
+ */
+ default List<Double> getGaugeValues(final String componentId, final String
name, final Map<String, String> attributes) {
+ throw new UnsupportedOperationException("This runner does not support
component metrics");
+ }
+
/**
* Returns the HTTP port on which the embedded Jetty server is listening,
or -1 if no server is running.
*
diff --git
a/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/RecordingComponentMetricReporter.java
b/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/RecordingComponentMetricReporter.java
new file mode 100644
index 00000000000..483340e02a1
--- /dev/null
+++
b/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/RecordingComponentMetricReporter.java
@@ -0,0 +1,93 @@
+/*
+ * 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.mock.connector.server;
+
+import org.apache.nifi.controller.metrics.ComponentMetricReporter;
+import org.apache.nifi.controller.metrics.CounterRecord;
+import org.apache.nifi.controller.metrics.GaugeRecord;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+class RecordingComponentMetricReporter implements ComponentMetricReporter {
+ private final ConcurrentMap<MetricKey, AtomicLong> counterValues = new
ConcurrentHashMap<>();
+ private final Queue<GaugeMeasurement> gaugeMeasurements = new
ConcurrentLinkedQueue<>();
+
+ @Override
+ public void recordGauge(final GaugeRecord gaugeRecord) {
+ final String componentId = gaugeRecord.componentMetricContext().id();
+ final MetricKey metricKey = new MetricKey(componentId,
gaugeRecord.name(), gaugeRecord.attributes());
+ gaugeMeasurements.add(new GaugeMeasurement(metricKey,
gaugeRecord.value()));
+ }
+
+ @Override
+ public void recordCounter(final CounterRecord counterRecord) {
+ final String componentId = counterRecord.componentMetricContext().id();
+ final MetricKey metricKey = new MetricKey(componentId,
counterRecord.name(), counterRecord.attributes());
+ counterValues.computeIfAbsent(metricKey, key -> new
AtomicLong()).addAndGet(counterRecord.value());
+ }
+
+ Long getCounterValue(final String componentId, final String name) {
+ Long counterValue = null;
+
+ for (final Map.Entry<MetricKey, AtomicLong> counterEntry :
counterValues.entrySet()) {
+ final MetricKey metricKey = counterEntry.getKey();
+ if (metricKey.componentId().equals(componentId) &&
metricKey.name().equals(name)) {
+ final long recorded = counterEntry.getValue().get();
+ counterValue = counterValue == null ? recorded : counterValue
+ recorded;
+ }
+ }
+
+ return counterValue;
+ }
+
+ Long getCounterValue(final String componentId, final String name, final
Map<String, String> attributes) {
+ final MetricKey metricKey = new MetricKey(componentId, name,
attributes);
+ final AtomicLong counterValue = counterValues.get(metricKey);
+ return counterValue == null ? null : counterValue.get();
+ }
+
+ List<Double> getGaugeValues(final String componentId, final String name) {
+ return gaugeMeasurements.stream()
+ .filter(measurement ->
measurement.key().componentId().equals(componentId))
+ .filter(measurement -> measurement.key().name().equals(name))
+ .map(GaugeMeasurement::value)
+ .toList();
+ }
+
+ List<Double> getGaugeValues(final String componentId, final String name,
final Map<String, String> attributes) {
+ final MetricKey metricKey = new MetricKey(componentId, name,
attributes);
+ return gaugeMeasurements.stream()
+ .filter(measurement -> measurement.key().equals(metricKey))
+ .map(GaugeMeasurement::value)
+ .toList();
+ }
+
+ private record MetricKey(String componentId, String name, Map<String,
String> attributes) {
+ private MetricKey {
+ attributes = Map.copyOf(attributes);
+ }
+ }
+
+ private record GaugeMeasurement(MetricKey key, double value) {
+ }
+}
diff --git
a/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/StandardConnectorMockServer.java
b/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/StandardConnectorMockServer.java
index ff4a18e6ae9..1c3f1454f1c 100644
---
a/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/StandardConnectorMockServer.java
+++
b/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/StandardConnectorMockServer.java
@@ -50,7 +50,6 @@ import org.apache.nifi.connectable.FlowFileTransferCounts;
import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.DecommissionTask;
import org.apache.nifi.controller.FlowController;
-import org.apache.nifi.controller.metrics.DefaultComponentMetricReporter;
import org.apache.nifi.controller.repository.FlowFileEventRepository;
import org.apache.nifi.controller.repository.metrics.RingBufferEventRepository;
import org.apache.nifi.controller.status.history.StatusHistoryDumpFactory;
@@ -115,6 +114,8 @@ public class StandardConnectorMockServer implements
ConnectorMockServer {
private static final Duration DEFAULT_STOP_TIMEOUT =
Duration.ofSeconds(60);
private static final long STOP_POLL_INTERVAL_MILLIS = 250L;
+ private final RecordingComponentMetricReporter componentMetricReporter =
new RecordingComponentMetricReporter();
+
private Bundle systemBundle;
private Set<Bundle> bundles;
private NiFiProperties nifiProperties;
@@ -150,7 +151,7 @@ public class StandardConnectorMockServer implements
ConnectorMockServer {
nifiProperties,
authorizer,
auditService,
- new DefaultComponentMetricReporter(),
+ componentMetricReporter,
new MockPropertyEncryptionProvider(),
bulletinRepository,
extensionManager,
@@ -491,6 +492,26 @@ public class StandardConnectorMockServer implements
ConnectorMockServer {
.toList();
}
+ @Override
+ public Long getCounterValue(final String componentId, final String name) {
+ return componentMetricReporter.getCounterValue(componentId, name);
+ }
+
+ @Override
+ public Long getCounterValue(final String componentId, final String name,
final Map<String, String> attributes) {
+ return componentMetricReporter.getCounterValue(componentId, name,
attributes);
+ }
+
+ @Override
+ public List<Double> getGaugeValues(final String componentId, final String
name) {
+ return componentMetricReporter.getGaugeValues(componentId, name);
+ }
+
+ @Override
+ public List<Double> getGaugeValues(final String componentId, final String
name, final Map<String, String> attributes) {
+ return componentMetricReporter.getGaugeValues(componentId, name,
attributes);
+ }
+
@Override
public VersionedExternalFlow getActiveFlowSnapshot() {
final FrameworkFlowContext activeFlowContext =
connectorNode.getActiveFlowContext();
diff --git
a/nifi-connector-mock-bundle/nifi-connector-mock-server/src/test/java/org/apache/nifi/mock/connector/server/RecordingComponentMetricReporterTest.java
b/nifi-connector-mock-bundle/nifi-connector-mock-server/src/test/java/org/apache/nifi/mock/connector/server/RecordingComponentMetricReporterTest.java
new file mode 100644
index 00000000000..2f3c5c929c9
--- /dev/null
+++
b/nifi-connector-mock-bundle/nifi-connector-mock-server/src/test/java/org/apache/nifi/mock/connector/server/RecordingComponentMetricReporterTest.java
@@ -0,0 +1,69 @@
+/*
+ * 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.mock.connector.server;
+
+import org.apache.nifi.controller.metrics.ComponentMetricContext;
+import org.apache.nifi.controller.metrics.CounterRecord;
+import org.apache.nifi.controller.metrics.GaugeRecord;
+import org.junit.jupiter.api.Test;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class RecordingComponentMetricReporterTest {
+ private static final String COMPONENT_ID = "component-id";
+ private static final String OTHER_COMPONENT_ID = "other-component-id";
+ private static final String METRIC_NAME = "metric";
+ private static final Map<String, String> ATTRIBUTES = Map.of("key",
"value");
+
+ private final RecordingComponentMetricReporter reporter = new
RecordingComponentMetricReporter();
+
+ @Test
+ void testCounterValues() {
+ reporter.recordCounter(new CounterRecord(METRIC_NAME, 2, Map.of(),
Instant.now(), getContext(COMPONENT_ID)));
+ reporter.recordCounter(new CounterRecord(METRIC_NAME, 3, ATTRIBUTES,
Instant.now(), getContext(COMPONENT_ID)));
+ reporter.recordCounter(new CounterRecord(METRIC_NAME, 7, ATTRIBUTES,
Instant.now(), getContext(OTHER_COMPONENT_ID)));
+
+ assertEquals(5L, reporter.getCounterValue(COMPONENT_ID, METRIC_NAME));
+ assertEquals(2L, reporter.getCounterValue(COMPONENT_ID, METRIC_NAME,
Map.of()));
+ assertEquals(3L, reporter.getCounterValue(COMPONENT_ID, METRIC_NAME,
ATTRIBUTES));
+ assertNull(reporter.getCounterValue(COMPONENT_ID, "unrecorded"));
+ assertNull(reporter.getCounterValue(COMPONENT_ID, METRIC_NAME,
Map.of("unrecorded", "attribute")));
+ }
+
+ @Test
+ void testGaugeValues() {
+ reporter.recordGauge(new GaugeRecord(METRIC_NAME, 2.0, Map.of(),
Instant.now(), getContext(COMPONENT_ID)));
+ reporter.recordGauge(new GaugeRecord(METRIC_NAME, 3.0, ATTRIBUTES,
Instant.now(), getContext(COMPONENT_ID)));
+ reporter.recordGauge(new GaugeRecord(METRIC_NAME, 7.0, ATTRIBUTES,
Instant.now(), getContext(OTHER_COMPONENT_ID)));
+
+ assertEquals(List.of(2.0, 3.0), reporter.getGaugeValues(COMPONENT_ID,
METRIC_NAME));
+ assertEquals(List.of(2.0), reporter.getGaugeValues(COMPONENT_ID,
METRIC_NAME, Map.of()));
+ assertEquals(List.of(3.0), reporter.getGaugeValues(COMPONENT_ID,
METRIC_NAME, ATTRIBUTES));
+ assertTrue(reporter.getGaugeValues(COMPONENT_ID,
"unrecorded").isEmpty());
+ assertTrue(reporter.getGaugeValues(COMPONENT_ID, METRIC_NAME,
Map.of("unrecorded", "attribute")).isEmpty());
+ }
+
+ private ComponentMetricContext getContext(final String componentId) {
+ return new ComponentMetricContext(componentId, "name", "type",
Map.of());
+ }
+}
diff --git
a/nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-integration-tests/src/test/java/org/apache/nifi/mock/connectors/tests/MockProcessorIT.java
b/nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-integration-tests/src/test/java/org/apache/nifi/mock/connectors/tests/MockProcessorIT.java
index 34680ed1cbc..c4fefe54e88 100644
---
a/nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-integration-tests/src/test/java/org/apache/nifi/mock/connectors/tests/MockProcessorIT.java
+++
b/nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-integration-tests/src/test/java/org/apache/nifi/mock/connectors/tests/MockProcessorIT.java
@@ -19,6 +19,7 @@ package org.apache.nifi.mock.connectors.tests;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.Validator;
+import org.apache.nifi.flow.VersionedProcessor;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.mock.connector.StandardConnectorTestRunner;
import org.apache.nifi.mock.connector.server.ConnectorTestRunner;
@@ -27,6 +28,7 @@ import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.metrics.CommitTiming;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.io.TempDir;
@@ -34,10 +36,19 @@ import org.junit.jupiter.api.io.TempDir;
import java.io.File;
import java.io.IOException;
import java.nio.file.Path;
+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 static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
public class MockProcessorIT {
+ private static final String COUNTER_NAME = "counter";
+ private static final String GAUGE_NAME = "gauge";
+ private static final Map<String, String> METRIC_ATTRIBUTES = Map.of("key",
"value");
@TempDir
private Path temporaryDirectory;
@@ -45,6 +56,9 @@ public class MockProcessorIT {
@Test
@Timeout(10)
public void testMockProcessor() throws IOException {
+ MockProcessor.invocationCounter.set(0L);
+ MockProcessor.metricsRecorded.set(false);
+
try (final ConnectorTestRunner runner = new
StandardConnectorTestRunner.Builder()
.narLibraryDirectory(new File("target/libDir"))
.connectorClassName("org.apache.nifi.mock.connectors.GenerateAndLog")
@@ -60,11 +74,26 @@ public class MockProcessorIT {
Thread.yield();
}
runner.stopConnector();
+
+ final VersionedProcessor processor =
runner.getActiveFlowSnapshot().getFlowContents().getProcessors().stream()
+ .filter(versionedProcessor ->
"UpdateAttribute".equals(versionedProcessor.getName()))
+ .findFirst()
+ .orElseThrow();
+ final String componentId = processor.getInstanceIdentifier();
+
+ assertEquals(5L, runner.getCounterValue(componentId,
COUNTER_NAME));
+ assertEquals(2L, runner.getCounterValue(componentId, COUNTER_NAME,
Map.of()));
+ assertEquals(3L, runner.getCounterValue(componentId, COUNTER_NAME,
METRIC_ATTRIBUTES));
+ assertNull(runner.getCounterValue(componentId, "unrecorded"));
+ assertEquals(List.of(2.0, 3.0), runner.getGaugeValues(componentId,
GAUGE_NAME));
+ assertEquals(List.of(2.0), runner.getGaugeValues(componentId,
GAUGE_NAME, Map.of()));
+ assertEquals(List.of(3.0), runner.getGaugeValues(componentId,
GAUGE_NAME, METRIC_ATTRIBUTES));
}
}
public static class MockProcessor extends AbstractProcessor {
private static final AtomicLong invocationCounter = new AtomicLong(0L);
+ private static final AtomicBoolean metricsRecorded = new
AtomicBoolean();
public static final Relationship REL_SUCCESS = new
Relationship.Builder()
.name("success")
@@ -88,6 +117,13 @@ public class MockProcessorIT {
public void onTrigger(final ProcessContext context, final
ProcessSession session) throws ProcessException {
invocationCounter.incrementAndGet();
+ if (metricsRecorded.compareAndSet(false, true)) {
+ session.adjustCounter(COUNTER_NAME, 2, Map.of(),
CommitTiming.NOW);
+ session.adjustCounter(COUNTER_NAME, 3, METRIC_ATTRIBUTES,
CommitTiming.NOW);
+ session.recordGauge(GAUGE_NAME, 2.0, Map.of(),
CommitTiming.NOW);
+ session.recordGauge(GAUGE_NAME, 3.0, METRIC_ATTRIBUTES,
CommitTiming.NOW);
+ }
+
final FlowFile flowFile = session.get();
if (flowFile != null) {
session.transfer(flowFile, REL_SUCCESS);
diff --git
a/nifi-connector-mock-bundle/nifi-connector-mock/src/main/java/org/apache/nifi/mock/connector/StandardConnectorTestRunner.java
b/nifi-connector-mock-bundle/nifi-connector-mock/src/main/java/org/apache/nifi/mock/connector/StandardConnectorTestRunner.java
index 313eb4c8f06..4e3bba4438f 100644
---
a/nifi-connector-mock-bundle/nifi-connector-mock/src/main/java/org/apache/nifi/mock/connector/StandardConnectorTestRunner.java
+++
b/nifi-connector-mock-bundle/nifi-connector-mock/src/main/java/org/apache/nifi/mock/connector/StandardConnectorTestRunner.java
@@ -248,6 +248,26 @@ public class StandardConnectorTestRunner implements
ConnectorTestRunner, Closeab
return mockServer.validate();
}
+ @Override
+ public Long getCounterValue(final String componentId, final String name) {
+ return mockServer.getCounterValue(componentId, name);
+ }
+
+ @Override
+ public Long getCounterValue(final String componentId, final String name,
final Map<String, String> attributes) {
+ return mockServer.getCounterValue(componentId, name, attributes);
+ }
+
+ @Override
+ public List<Double> getGaugeValues(final String componentId, final String
name) {
+ return mockServer.getGaugeValues(componentId, name);
+ }
+
+ @Override
+ public List<Double> getGaugeValues(final String componentId, final String
name, final Map<String, String> attributes) {
+ return mockServer.getGaugeValues(componentId, name, attributes);
+ }
+
@Override
public int getHttpPort() {
return mockServer.getHttpPort();
