private void convertConfigToCoordinatorMessage(Config config) {
    try {
      for (Map.Entry<String, String> configPair : config.entrySet()) {
        byte[] keyBytes;
        byte[] messgeBytes;
        if (configPair.getKey().startsWith(CHANGELOGPREFIX)) {
          String[] changelogInfo = configPair.getKey().split(":");
          String changeLogPartition = configPair.getValue();
          SetChangelogMapping changelogMapping = new SetChangelogMapping(changelogInfo[1], changelogInfo[2], Integer.parseInt(changeLogPartition));
          keyBytes = MAPPER.writeValueAsString(changelogMapping.getKeyArray()).getBytes("UTF-8");
          messgeBytes = MAPPER.writeValueAsString(changelogMapping.getMessageMap()).getBytes("UTF-8");
        } else {
          SetConfig setConfig = new SetConfig("source", configPair.getKey(), configPair.getValue());
          keyBytes = MAPPER.writeValueAsString(setConfig.getKeyArray()).getBytes("UTF-8");
          messgeBytes = MAPPER.writeValueAsString(setConfig.getMessageMap()).getBytes("UTF-8");
        }
        // The ssp here is the coordinator ssp (which is always fixed) and not the task ssp.
        put(systemStreamPartition, new IncomingMessageEnvelope(systemStreamPartition, "", keyBytes, messgeBytes));
      }
      setIsAtHead(systemStreamPartition, true);
    } catch (Exception e) {
      throw new SamzaException(e);
    }
  }
A mock SystemConsumer that pretends to be a coordinator stream. The mock will
take all configs given to it, and put them into the coordinator stream's
SystemStreamPartition. This is useful in cases where config needs to be
quickly passed from a unit test into the JobCoordinator.
samza-core/src/test/java/org/apache/samza/coordinator/stream/MockCoordinatorStreamWrappedConsumer.java