Metrics { public void offsetHighwaterMark(int partition, long offset) { highwaterMarks.computeIfAbsent(partition, this::createGauge).set(offset); } @Autowired  Metrics(MetricRegistry registry); void consumedBytes(long byteCount); void uploadedBytes(long byteCount); void uploadedEvents(long eventCount); void offsetHighwaterMark(int partition, long offset);  }
@Test public void offsetHighwaterMark() { ArgumentCaptor<SettableGauge> gauge = ArgumentCaptor.forClass(SettableGauge.class); when(metricRegistry.register(eq("partition.0.highwater-mark"), gauge.capture())).thenReturn(null); new Metrics(metricRegistry).offsetHighwaterMark(0, 1L); verify(metricRegistry).register(eq("partition.0.highwater-mark"), gauge.capture()); assertThat(gauge.getValue().getValue(), is(1L)); }