### Question:
CloudwatchNiFiClusterMetricsReporter extends AbstractNiFiClusterMetricsReporter { List<MetricDatum> collectMeasurements(Date now, SystemMetricsSnapshot snapshot, List<Dimension> dimensions) { List<MetricDatum> toCloudwatch = new ArrayList<>(); if (collectsMemory) { getMetrics("System Memory", snapshot.getMachineMemory(), now, dimensions, toCloudwatch); } if (collectsJVMMetrics) { getMetrics("System JVM", snapshot.getJvmMetrics(), now, dimensions, toCloudwatch); } snapshot.getProcessGroupSnapshots().forEach((groupData) -> getMetrics(groupData.getProcessGroupName(), groupData.valuesAsMap(), now, dimensions, toCloudwatch) ); snapshot.getRemoteProcessGroupSnapshots().forEach((groupData) -> getMetrics(groupData.getRemoteProcessGroupName(), groupData.valuesAsMap(), now, dimensions, toCloudwatch) ); snapshot.getProcessorSnapshots().forEach((processorData) -> getMetrics(processorData.getProcessorName(), processorData.valuesAsMap(), now, dimensions, toCloudwatch) ); snapshot.getConnectionSnapshots().forEach((connectionData) -> getMetrics(connectionData.getConnectionName(), connectionData.valuesAsMap(), now, dimensions, toCloudwatch) ); snapshot.getInputPortSnapshots().forEach((portData) -> getMetrics(portData.getInputPortName(), portData.valuesAsMap(), now, dimensions, toCloudwatch) ); snapshot.getOutputPortSnapshots().forEach((portData) -> getMetrics(portData.getInputPortName(), portData.valuesAsMap(), now, dimensions, toCloudwatch) ); snapshot.getDiskMetrics().entrySet().forEach((fileData) -> getMetrics(fileData.getKey().getAbsolutePath(), fileData.getValue(), now, dimensions, toCloudwatch) ); return toCloudwatch; }  @Override final List<PropertyDescriptor> getSupportedPropertyDescriptors(); @OnScheduled void startup(ConfigurationContext context); static final PropertyDescriptor CREDENTIALS_FILE; static final PropertyDescriptor ACCESS_KEY; static final PropertyDescriptor SECRET_KEY; }### Answer:
@Test public void testCollectMeasurements() throws InitializationException { CloudwatchNiFiClusterMetricsReporter metricsCloudwatchReporter = new CloudwatchNiFiClusterMetricsReporter(); ProcessGroupStatusMetric processGroupStatusMetric = new ProcessGroupStatusMetric() .setId("id") .setProcessGroupName("root") .setActiveThreadCount(15.0) .setBytesRead(1.4) .setBytesReceived(4.) .setBytesSent(45.) .setBytesWritten(2387.) .setFlowFilesReceived(3425.) .setFlowFilesSent(3425.) .setFlowFilesTransferred(4243.) .setQueued(34521.) .setQueuedContentSize(487652.); Map<String, Object> memory = new HashMap<>(); memory.put("mem", 1.2); Map<String, Object> jvm = new HashMap<>(); jvm.put("free", 1.2); SystemMetricsSnapshot metrics = new SystemMetricsSnapshot() .setClusterNodeIdentifier("clusterId") .setMachineMemory(memory) .setJvmMetrics(jvm) .setRootProcessGroupSnapshot(processGroupStatusMetric) .setProcessGroupSnapshots(Arrays.asList(processGroupStatusMetric)); List<Dimension> dimensions = new ArrayList<>(); List<MetricDatum> cloudwatchMetrics = metricsCloudwatchReporter.collectMeasurements(new Date(), metrics, dimensions); assertEquals(10, cloudwatchMetrics.size()); metricsCloudwatchReporter.collectsJVMMetrics = true; metricsCloudwatchReporter.collectsMemory = true; cloudwatchMetrics = metricsCloudwatchReporter.collectMeasurements(new Date(), metrics, dimensions); assertEquals(12, cloudwatchMetrics.size()); }