public static Validator mapWithIntKeyDoubleValue() {
    return (name, val) -> {
      if (!(val instanceof String)) {
        throw new ConfigException(name, val, "Must be a string");
      }

      final String str = (String) val;
      final Map<String, String> map = KsqlConfig.parseStringAsMap(name, str);
      map.forEach((keyStr, valueStr) -> {
        try {
          Integer.parseInt(keyStr);
        } catch (NumberFormatException e) {
          throw new ConfigException(name, keyStr, "Not an int");
        }
        try {
          Double.parseDouble(valueStr);
        } catch (NumberFormatException e) {
          throw new ConfigException(name, valueStr, "Not a double");
        }
      });
    };
  }
@Test
  public void shouldThrowOnBadIntDoubleValueInMap() {
    // Given:
    final Validator validator = ConfigValidators.mapWithIntKeyDoubleValue();

    // When:
    final Exception e = assertThrows(
        ConfigException.class,
        () -> validator.ensureValid("propName", "1:abc")
    );
    final Exception e2 = assertThrows(
        ConfigException.class,
        () -> validator.ensureValid("propName", "abc:1.2")
    );

    // Then:
    assertThat(e.getMessage(),
        containsString("Invalid value abc for configuration propName: Not a double"));
    assertThat(e2.getMessage(),
        containsString("Invalid value abc for configuration propName: Not an int"));
  }