public static Map<String, MetricsSystem.SinkProperties> parseSinkProperties(
      Properties metricsProperties) {
    Map<String, MetricsSystem.SinkProperties> propertiesMap = new HashMap<>();
    // e.g: "sink.graphite.class"="org.apache.spark.metrics.sink.ConsoleSink"
    Enumeration<?> valueEnumeration = metricsProperties.propertyNames();
    while (valueEnumeration.hasMoreElements()) {
      String key = (String) valueEnumeration.nextElement();
      int firstDotIndex = StringUtils.ordinalIndexOf(key, ".", 1);
      int secondDotIndex = StringUtils.ordinalIndexOf(key, ".", 2);
      if (key.startsWith(SINK)) {
        String shortName = key.substring(firstDotIndex + 1, secondDotIndex);
        MetricsSystem.SinkProperties sinkProperties =
            propertiesMap.getOrDefault(shortName, new MetricsSystem.SinkProperties());
        if (key.endsWith(CLASS)) {
          sinkProperties.setClassName(metricsProperties.getProperty(key));
        } else {
          sinkProperties
              .getProperties()
              .put(key.substring(secondDotIndex + 1), metricsProperties.getProperty(key));
        }
        propertiesMap.put(shortName, sinkProperties);
      }
    }
    sinkPropertiesSanityCheck(propertiesMap);
    return propertiesMap;
  }
@Test
  void testMetricsSystemFailFastWithNotFoundClassName() {
    Properties properties = new Properties();
    properties.put("sink.console.class", "org.apache.spark.metrics.sink.FooSink");
    RuntimeException e =
        assertThrows(RuntimeException.class, () -> parseSinkProperties(properties));
    Assertions.assertEquals(
        "Fail to find class org.apache.spark.metrics.sink.FooSink", e.getMessage());
  }