@Test
public void testFederatedWorkerEndToEnd() throws Exception {

  class WorkerStarterRunnable implements Runnable {
    private final WorkerStarter starter;
    public WorkerStarterRunnable(WorkerConf workerConf) {
      starter = new WorkerStarter(workerConf);
    }
    @Override
    public void run() {
      try {
        LOGGER.info("Starting WorkerStarter");
        starter.run();
      } catch (Exception e) {
        LOGGER.error("WorkerStarter failed", e);
      }
    }
    public void shutdown() {
      starter.shutdown();
    }
  }
  // create new topic for testing
  String topicName1 = "tessFederatedWorkerEndToEnd1";
  String topicName2 = "tessFederatedWorkerEndToEnd2";
  String topicName3 = "tessFederatedWorkerEndToEnd3";
  KafkaStarterUtils.createTopic(topicName1, numberOfPartitions, srcCluster1ZK, "2");
  KafkaStarterUtils.createTopic(topicName2, numberOfPartitions, srcCluster1ZK, "2");
  KafkaStarterUtils.createTopic(topicName3, numberOfPartitions, srcCluster2ZK, "1");
  KafkaStarterUtils.createTopic(topicName1, numberOfPartitions, dstClusterZK, "1");
  KafkaStarterUtils.createTopic(topicName2, numberOfPartitions, dstClusterZK, "1");
  KafkaStarterUtils.createTopic(topicName3, numberOfPartitions, dstClusterZK, "1");
  // init zk client for zk cluster 1
  ZkClient commitZkClient = ZkUtils.createZkClient(
      srcCluster1ZK,
      10000,
      10000);
  WorkerConf workerConf = TestUtils.initWorkerConf();
  Properties helixProps = WorkerUtils.loadAndValidateHelixProps(workerConf.getHelixConfigFile());
  String instanceId = helixProps.getProperty(Constants.HELIX_INSTANCE_ID, null);
  Assert.assertNotNull(instanceId, String
      .format("failed to find property %s in configuration file %s", Constants.HELIX_INSTANCE_ID,
          workerConf.getHelixConfigFile()));
  // prepare helix cluster
  String route1 = String.format("%s-%s-0", TestUtils.SRC_CLUSTER, TestUtils.DST_CLUSTER);
  String route1ForHelix = String.format("@%s@%s", TestUtils.SRC_CLUSTER, TestUtils.DST_CLUSTER);
  String route2 = String.format("%s-%s-0", TestUtils.SRC_CLUSTER_2, TestUtils.DST_CLUSTER);
  String route2ForHelix = String.format("@%s@%s", TestUtils.SRC_CLUSTER_2, TestUtils.DST_CLUSTER);
  ZKHelixAdmin helixAdmin = TestUtils.initHelixClustersForWorkerTest(helixProps, route1, route2);
  String deployment = helixProps.getProperty("federated.deployment.name");
  String managerHelixClusterName = WorkerUtils.getManagerWorkerHelixClusterName(deployment);
  String controllerHelixClusterName = WorkerUtils.getControllerWorkerHelixClusterName(route1);
  String controller2HelixClusterName = WorkerUtils.getControllerWorkerHelixClusterName(route2);
  Thread.sleep(1000);
  WorkerStarterRunnable runnable = new WorkerStarterRunnable(workerConf);
  Thread workerThread = new Thread(runnable);
  workerThread.start();
  // assign worker to route1
  TestUtils.updateRouteWithValidation(managerHelixClusterName, route1ForHelix, instanceId, helixAdmin, "ONLINE");
  TestUtils.updateTopicWithValidation(controllerHelixClusterName, topicName1, Arrays.asList(0, 1),
      Arrays.asList("0"), helixAdmin, "ONLINE");
  TestUtils.produceMessages(srcCluster1BootstrapServer, topicName1, 20);
  List<ConsumerRecord<Byte[], Byte[]>> records = TestUtils
      .consumeMessage(dstBootstrapServer, topicName1, 6000);
  Assert.assertEquals(records.size(), 20);
  TestUtils.updateTopicWithValidation(controllerHelixClusterName, topicName1, Arrays.asList(0, 1),
      Arrays.asList("0"), helixAdmin, "OFFLINE");
  TestUtils.produceMessages(srcCluster1BootstrapServer, topicName1, 20);
  records = TestUtils.consumeMessage(dstBootstrapServer, topicName1, 6000);
  Assert.assertEquals(records.size(), 0);
  TestUtils.updateRouteWithValidation(managerHelixClusterName, route1ForHelix, instanceId, helixAdmin, "OFFLINE");
  String offset = commitZkClient
      .readData("/consumers/ureplicator-cluster1-cluster2/offsets/" + topicName1 + "/0")
      .toString();
  Assert.assertEquals(offset, "10");
  // assign worker to route1
  TestUtils.updateRouteWithValidation(managerHelixClusterName, route1ForHelix, instanceId, helixAdmin, "ONLINE");
  TestUtils.updateTopicWithValidation(controllerHelixClusterName, topicName1, Arrays.asList(0, 1),
      Arrays.asList("0"), helixAdmin, "ONLINE");
  records = TestUtils.consumeMessage(dstBootstrapServer, topicName1, 6000);
  Assert.assertEquals(records.size(), 20);
  TestUtils.updateRouteWithValidation(managerHelixClusterName, route1ForHelix, instanceId, helixAdmin, "OFFLINE");
  TestUtils.updateRouteWithValidation(managerHelixClusterName, route2ForHelix, instanceId, helixAdmin, "ONLINE");
  TestUtils.updateTopicWithValidation(controller2HelixClusterName, topicName3, Arrays.asList(0, 1),
      Arrays.asList("0"), helixAdmin, "ONLINE");
  TestUtils.produceMessages(srcCluster2BootstrapServer, topicName3, 20);
  records = TestUtils.consumeMessage(dstBootstrapServer, topicName3, 6000);
  Assert.assertEquals(records.size(), 20);
  runnable.shutdown();
  offset = commitZkClient
      .readData("/consumers/ureplicator-cluster1-cluster2/offsets/" + topicName1 + "/0")
      .toString();
  Assert.assertEquals(offset, "20");
}