public static Map<String, Object> coerceTypes(
      final Map<String, Object> streamsProperties,
      final boolean ignoreUnresolved
  ) {
    if (streamsProperties == null) {
      return Collections.emptyMap();
    }

    final Map<String, Object> validated = new HashMap<>(streamsProperties.size());
    for (final Map.Entry<String, Object> e : streamsProperties.entrySet()) {
      try {
        validated.put(e.getKey(), coerceType(e.getKey(), e.getValue()));
      } catch (final PropertyNotFoundException p) {
        if (ignoreUnresolved) {
          validated.put(e.getKey(), e.getValue());
        } else {
          throw p;
        }
      }
    }
    return validated;
  }
@Test
  public void shouldCoerceTypes() {
    // given/when:
    final Map<String, Object> coerced = PropertiesUtil.coerceTypes(ImmutableMap.of(
        "ksql.internal.topic.replicas", 3L,
        "cache.max.bytes.buffering", "0"
    ), false);

    // then:
    assertThat(coerced.get("ksql.internal.topic.replicas"), instanceOf(Short.class));
    assertThat(coerced.get("ksql.internal.topic.replicas"), equalTo((short) 3));
    assertThat(coerced.get("cache.max.bytes.buffering"), instanceOf(Long.class));
    assertThat(coerced.get("cache.max.bytes.buffering"), equalTo(0L));
  }