public static synchronized BigInteger getMaxTaskUsage(final Metrics metricRegistry) {
    final Collection<KafkaMetric> taskMetrics = metricRegistry
        .metrics()
        .entrySet()
        .stream()
        .filter(e -> e.getKey().name().contains(TASK_STORAGE_USED_BYTES))
        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))
        .values();
    final Optional<BigInteger> maxOfTaskMetrics = taskMetrics
        .stream()
        .map(e -> (BigInteger) e.metricValue())
        .reduce(BigInteger::max);
    return maxOfTaskMetrics.orElse(BigInteger.ZERO);
  }
@Test
  public void shouldRecordMaxTaskUsageWithNoTasks() {
    // Given:
    when(metrics.metrics()).thenReturn(Collections.emptyMap());

    // When:

    // Then:
    BigInteger maxVal = StorageUtilizationMetricsReporter.getMaxTaskUsage(metrics);
    assertEquals(maxVal, BigInteger.valueOf(0));
  }