docs: protected static Map<String, String> getComponentConfigMap(TopologyAPI.Topology topology,
                                                               String key) throws RuntimeException {
      List<TopologyAPI.Config.KeyValue> topologyConfig = topology.getTopologyConfig().getKvsList();
      Map<String, String> configMap = new HashMap<>();
  
      // Get the set of component names to make sure the config only specifies valid component name
      Set<String> componentNames = getComponentParallelism(topology).keySet();
  
      // Parse the config value
      String mapStr = getConfigWithDefault(topologyConfig, key, (String) null);
      if (mapStr != null) {
        String[] mapTokens = mapStr.split(",");
        // Each token should be in this format: component:value
        for (String token : mapTokens) {
          if (token.trim().isEmpty()) {
            continue;
          }
          String[] componentAndValue = token.split(":");
          if (componentAndValue.length != 2) {
            throw new RuntimeException("Malformed component config " + key);
          }
          if (!componentNames.contains(componentAndValue[0])) {
            throw new RuntimeException("Invalid component. " + componentAndValue[0] + " not found");
          }
          configMap.put(componentAndValue[0], componentAndValue[1]);
        }
      }
      return configMap;
    }