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); initializeConsumerStream(); initializeTopicObserver(); initializeMetricsReporter(srcCluster, dstCluster, routeId, federatedDeploymentName); additionalConfigs(srcCluster, dstCluster); messageTransformer = createMessageTransformer(); checkpointManager = createCheckpointManager(); String clientId = "ureplicator"; if (workerConf.getFederatedEnabled()) { clientId = String.format("ureplicator-%s-%s-%s", srcCluster, dstCluster, routeId); } consumerProps.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, clientId); 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(); } WorkerInstance(WorkerConf workerConf); boolean isRunning(); void start(String srcCluster, String dstCluster, String routeId,
      String federatedDeploymentName); IMessageTransformer createMessageTransformer(); ICheckPointManager createCheckpointManager(); IConsumerFetcherManager createFetcherManager(); ProducerManager createProducerManager(); void addTopicPartition(String topic, int partition); void addTopicPartition(String topic, int partition, Long startingOffset,
      Long endingOffset, String dstTopic); void deleteTopicPartition(String topic, int partition); void cleanShutdown(); void cleanShutdown(boolean force); void additionalConfigs(String srcCluster, String dstCluster); void setMessageRatePerSecond(Double messageRatePerSecond);  }
@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(); } } 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"); 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())); 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(); 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"); 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"); }