void recordLatency(String node, long requestLatencyMs) {
        fetchLatency.record(requestLatencyMs);
        if (!node.isEmpty()) {
            String nodeTimeName = "node-" + node + ".latency";
            Sensor nodeRequestTime = this.metrics.getSensor(nodeTimeName);
            if (nodeRequestTime != null)
                nodeRequestTime.record(requestLatencyMs);
        }
    }
@Test
    public void testNodeLatency() {
        String connectionId = "0";
        MetricName nodeLatencyAvg = metrics.metricName("request-latency-avg", "group");
        MetricName nodeLatencyMax = metrics.metricName("request-latency-max", "group");
        registerNodeLatencyMetric(connectionId, nodeLatencyAvg, nodeLatencyMax);

        metricsManager.recordLatency(connectionId, 123);
        time.sleep(metrics.config().timeWindowMs() + 1);
        metricsManager.recordLatency(connectionId, 456);

        assertEquals(289.5, metricValue(metricsRegistry.fetchLatencyAvg), EPSILON);
        assertEquals(456, metricValue(metricsRegistry.fetchLatencyMax), EPSILON);

        assertEquals(289.5, metricValue(nodeLatencyAvg), EPSILON);
        assertEquals(456, metricValue(nodeLatencyMax), EPSILON);

        // Record metric against another node.
        metricsManager.recordLatency("1", 501);

        assertEquals(360, metricValue(metricsRegistry.fetchLatencyAvg), EPSILON);
        assertEquals(501, metricValue(metricsRegistry.fetchLatencyMax), EPSILON);
        // Node specific metric should not be affected.
        assertEquals(289.5, metricValue(nodeLatencyAvg), EPSILON);
        assertEquals(456, metricValue(nodeLatencyMax), EPSILON);
    }