protected CoordinatorStreamSystemConsumer initializeCoordinatorStreamConsumer(JobInstance jobInstance) {
    Config coordinatorSystemConfig = getCoordinatorSystemConfig(jobInstance);
    LOG.debug("Using config: {} to create coordinatorStream consumer.", coordinatorSystemConfig);
    CoordinatorStreamSystemConsumer consumer = new CoordinatorStreamSystemConsumer(coordinatorSystemConfig, METRICS_REGISTRY);
    LOG.debug("Registering coordinator system stream consumer.");
    consumer.register();
    LOG.debug("Starting coordinator system stream consumer.");
    consumer.start();
    LOG.debug("Bootstrapping coordinator system stream consumer.");
    consumer.bootstrap();
    return consumer;
  }
Initialize {@link CoordinatorStreamSystemConsumer} based upon {@link JobInstance} parameter.
@param jobInstance the job instance to get CoordinatorStreamSystemConsumer for.
@return built and initialized CoordinatorStreamSystemConsumer.
samza-rest/src/main/java/org/apache/samza/rest/proxy/task/SamzaTaskProxy.java