pvillard31 commented on code in PR #11398:
URL: https://github.com/apache/nifi/pull/11398#discussion_r3532201666
##########
nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java:
##########
@@ -111,8 +111,8 @@
import org.apache.nifi.controller.flow.FlowManager;
import org.apache.nifi.controller.label.Label;
import org.apache.nifi.controller.leader.election.LeaderElectionManager;
-import org.apache.nifi.controller.repository.FlowFileEvent;
import org.apache.nifi.controller.repository.FlowFileEventRepository;
+import org.apache.nifi.controller.metrics.ProcessSessionEvent;
Review Comment:
Can the new org.apache.nifi.controller.metrics.ProcessSessionEvent import be
moved above the org.apache.nifi.controller.repository imports so the Validate
checkstyle CustomImportOrder check passes?
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java:
##########
@@ -1829,23 +1838,47 @@ private void formatNanos(final long nanos, final
StringBuilder sb) {
}
private void incrementConnectionInputCounts(final Connection connection,
final RepositoryRecord record) {
- incrementConnectionInputCounts(connection.getIdentifier(), 1,
record.getCurrent().getSize());
+ incrementConnectionInputCounts(connection, 1,
record.getCurrent().getSize());
}
- private void incrementConnectionInputCounts(final String connectionId,
final int flowFileCount, final long bytes) {
- final StandardFlowFileEvent connectionEvent =
connectionCounts.computeIfAbsent(connectionId, id -> new
StandardFlowFileEvent());
- connectionEvent.setContentSizeIn(connectionEvent.getContentSizeIn() +
bytes);
- connectionEvent.setFlowFilesIn(connectionEvent.getFlowFilesIn() +
flowFileCount);
+ private void incrementConnectionInputCounts(final Connection connection,
final int flowFileCount, final long bytes) {
+ final String connectionId = connection.getIdentifier();
+ cacheConnectionMetricContext(connection);
+ final ProcessSessionEventBuilder connectionEvent =
connectionCounts.computeIfAbsent(
+ connectionId, id ->
ProcessSessionEventBuilder.forComponent(getConnectionMetricContext(connectionId)));
+ connectionEvent.addFlowFilesIn(flowFileCount).addContentSizeIn(bytes);
}
private void incrementConnectionOutputCounts(final Connection connection,
final FlowFileRecord record) {
- incrementConnectionOutputCounts(connection.getIdentifier(), 1,
record.getSize());
+ incrementConnectionOutputCounts(connection, 1, record.getSize());
+ }
+
+ private void incrementConnectionOutputCounts(final Connection connection,
final int flowFileCount, final long bytes) {
+ final String connectionId = connection.getIdentifier();
+ cacheConnectionMetricContext(connection);
+ final ProcessSessionEventBuilder connectionEvent =
connectionCounts.computeIfAbsent(
+ connectionId, id ->
ProcessSessionEventBuilder.forComponent(getConnectionMetricContext(connectionId)));
+
connectionEvent.addFlowFilesOut(flowFileCount).addContentSizeOut(bytes);
}
private void incrementConnectionOutputCounts(final String connectionId,
final int flowFileCount, final long bytes) {
- final StandardFlowFileEvent connectionEvent =
connectionCounts.computeIfAbsent(connectionId, id -> new
StandardFlowFileEvent());
- connectionEvent.setContentSizeOut(connectionEvent.getContentSizeOut()
+ bytes);
- connectionEvent.setFlowFilesOut(connectionEvent.getFlowFilesOut() +
flowFileCount);
+ final ProcessSessionEventBuilder connectionEvent =
connectionCounts.computeIfAbsent(
+ connectionId, id ->
ProcessSessionEventBuilder.forComponent(getConnectionMetricContext(connectionId)));
Review Comment:
In the id-only incrementConnectionOutputCounts used by migrate(), is the
fallback context (name equal to id, no group attributes) intended for the
destination session that never cached the connection?
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/StandardProcessSessionIT.java:
##########
@@ -177,10 +178,10 @@ public void setup() throws IOException {
resourceClaimManager = new StandardResourceClaimManager();
System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH,
StandardProcessSessionIT.class.getResource("/conf/nifi.properties").getFile());
+ componentMetricReporter = mock(ComponentMetricReporter.class);
Review Comment:
Could a test verify that ComponentMetricReporter.recordProcessSessionEvent()
is invoked on commit, so the new reporting path is locked in?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
