@Override
  public SystemConsumer getConsumer(String systemName, Config config, MetricsRegistry registry) {
    if (useCachedConsumer && mockConsumer != null) {
      return mockConsumer;
    }

    String jobName = config.get("job.name");
    String jobId = config.get("job.id");
    if (jobName == null) {
      throw new ConfigException("Must define job.name.");
    }
    if (jobId == null) {
      jobId = "1";
    }
    String streamName = CoordinatorStreamUtil.getCoordinatorStreamName(jobName, jobId);
    SystemStreamPartition systemStreamPartition = new SystemStreamPartition(systemName, streamName, new Partition(0));
    mockConsumer = new MockCoordinatorStreamWrappedConsumer(systemStreamPartition, config);
    mockConsumer.register(systemStreamPartition, "0");
    return mockConsumer;
  }
Returns a consumer that sends all configs to the coordinator stream.

@param config Along with the configs, you can pass checkpoints and changelog stream messages into the stream.
              The expected pattern is cp:source:taskname -> ssp,offset for checkpoint (Use sspToString util)
              ch:source:taskname -> changelogPartition for changelog
              Everything else is processed as normal config
samza-core/src/test/java/org/apache/samza/coordinator/stream/MockCoordinatorStreamSystemFactory.java