KAFKA-19888: Clamp negative values in coordinator histograms (#20986)

The coordinator runtime and group coordinator currently use wall clock
time to measure durations for metrics. When the system clock goes
backwards due to time adjustments, we attempt to record negative
durations for metrics, which throws an ArrayIndexOutOfBoundsException
exception. This causes request processing and partition loading to fail
while the clock is being adjusted. If partition loading fails, the group
the coordinator for that partition becomes unavailable until the broker
is restarted or leadership changes again.

To address this, we clamp negative durations to zero in histograms
instead of throwing ArrayIndexOutOfBoundsExceptions. We will move
towards using a monotonic clock for metrics in future work.

Reviewers: David Jacot <[email protected]>

Author: squah-confluent

coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/HdrHistogram.java

@@ -92,12 +92,12 @@ private Histogram latestHistogram(long now) {
     }
 
     /**
-     * Writes to the histogram. Caps recording to highestTrackableValue
+     * Writes to the histogram. Caps recording between 0 and highestTrackableValue.
      *
-     * @param value The value to be recorded. Cannot be negative.
+     * @param value The value to be recorded.
      */
     public void record(long value) {
-        recorder.recordValue(Math.min(value, highestTrackableValue));
+        recorder.recordValue(Math.min(Math.max(value, 0), highestTrackableValue));
     }
 
     /**

coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/HdrHistogramTest.java

@@ -216,10 +216,14 @@ public void testLatestHistogramRace() throws InterruptedException, ExecutionExce
 
     @Test
     public void testRecordLimit() {
+        long now = System.currentTimeMillis();
         long highestTrackableValue = 10L;
         HdrHistogram hdrHistogram = new HdrHistogram(10L, highestTrackableValue, 3);
 
         hdrHistogram.record(highestTrackableValue + 1000L);
-        assertEquals(highestTrackableValue, hdrHistogram.max(System.currentTimeMillis()));
+        assertEquals(highestTrackableValue, hdrHistogram.max(now));
+
+        hdrHistogram.record(-50L);
+        assertEquals(0, hdrHistogram.max(now + 1000L));
     }
 }