public static Map<String, String> getConfigMapWithPrefix(Map<String, String> batchConfigMap, String prefix) {
    Map<String, String> props = new HashMap<>();
    if (!prefix.endsWith(DOT_SEPARATOR)) {
      prefix = prefix + DOT_SEPARATOR;
    }
    for (String configKey : batchConfigMap.keySet()) {
      if (configKey.startsWith(prefix)) {
        String[] splits = configKey.split(prefix, 2);
        if (splits.length > 1) {
          props.put(splits[1], batchConfigMap.get(configKey));
        }
      }
    }
    return props;
  }
@Test
  public void testGetConfigMapWithPrefix() {
    Map<String, String> testMap = ImmutableMap.of("k1", "v1", "k1.k2", "v2", "k1.k3", "v3", "k4", "v4");
    Assert.assertEquals(2, IngestionConfigUtils.getConfigMapWithPrefix(testMap, "k1").size());
    Assert.assertEquals(2, IngestionConfigUtils.getConfigMapWithPrefix(testMap, "k1.").size());
  }