class WorkerInstance {

  public void start(String srcCluster, String dstCluster, String routeId,
      String federatedDeploymentName) {
    if (!isRunning.compareAndSet(false, true)) {
      LOGGER.error(
          "Instance already running, srcCluster: {}, dstCluster:{}",
          this.srcCluster, this.dstCluster);
      throw new InternalError(String
          .format("Instance already running, srcCluster: %s, dstCluster:%s", this.srcCluster,
              this.dstCluster));
    }
    isShuttingDown.set(false);
    this.srcCluster = srcCluster;
    this.dstCluster = dstCluster;
    initializeProperties(srcCluster, dstCluster);
    // Init blocking queue
    initializeConsumerStream();
    initializeTopicObserver();
    initializeMetricsReporter(srcCluster, dstCluster, routeId, federatedDeploymentName);
    additionalConfigs(srcCluster, dstCluster);

    messageTransformer = createMessageTransformer();

    checkpointManager = createCheckpointManager();

    // set client id prefix
    String clientId = "ureplicator";
    if (workerConf.getFederatedEnabled()) {
      clientId = String.format("ureplicator-%s-%s-%s", srcCluster, dstCluster, routeId);
    }
    consumerProps.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, clientId);
    // set consumer group to ureplicator it does not exists
    String consumerGroup = consumerProps.getProperty(ConsumerConfig.GROUP_ID_CONFIG, "ureplicator");
    consumerProps.setProperty(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup);
    fetcherManager = createFetcherManager();
    fetcherManager.start();

    producerManager = createProducerManager();
    producerManager.start();
    registerMetrics();
  }

  public  WorkerInstance(WorkerConf workerConf);

}