@Test(enabled = false) public void testNodeUpRebalanceTwoDatastreamsSimpleStrategy() throws Exception { _datastreamCluster = initializeTestDatastreamServerWithFileConnector(2, LOAD_BALANCING_STRATEGY_FACTORY); _datastreamCluster.startupServer(0); List<DatastreamServer> servers = _datastreamCluster.getAllDatastreamServers(); Assert.assertEquals(servers.size(), 2); Assert.assertNotNull(servers.get(0)); DatastreamServer server1 = servers.get(0); Path tempFile1 = Files.createTempFile("testFile1", ""); String fileName1 = tempFile1.toAbsolutePath().toString(); Path tempFile2 = Files.createTempFile("testFile2", ""); String fileName2 = tempFile2.toAbsolutePath().toString(); Datastream fileDatastream1 = createFileDatastream(fileName1, 1); Datastream fileDatastream2 = createFileDatastream(fileName2, 1); int totalEvents = 10; List<String> eventsWritten1 = TestUtils.generateStrings(totalEvents); List<String> eventsWritten2 = TestUtils.generateStrings(totalEvents); FileUtils.writeLines(new File(fileName1), eventsWritten1); FileUtils.writeLines(new File(fileName2), eventsWritten2); List<String> eventsReceived1 = readFileDatastreamEvents(fileDatastream1, totalEvents); List<String> eventsReceived2 = readFileDatastreamEvents(fileDatastream2, totalEvents); LOG.info("(1) Events Received " + eventsReceived1); LOG.info("(1) Events Written to file " + eventsWritten1); LOG.info("(2) Events Received " + eventsReceived2); LOG.info("(2) Events Written to file " + eventsWritten2); Assert.assertTrue(eventsReceived1.containsAll(eventsWritten1)); Assert.assertTrue(eventsReceived2.containsAll(eventsWritten2)); String cluster = _datastreamCluster.getDatastreamServerProperties().get(0).getProperty(CONFIG_CLUSTER_NAME); ZkClient zkclient = new ZkClient(_datastreamCluster.getZkConnection()); String instance1 = server1.getCoordinator().getInstanceName(); String assignmentPath = KeyBuilder.instanceAssignments(cluster, instance1); List<String> assignments = zkclient.getChildren(assignmentPath); Assert.assertEquals(assignments.size(), 2); _datastreamCluster.startupServer(1); DatastreamServer server2 = servers.get(1); Assert.assertNotNull(server2); String instancesPath = KeyBuilder.liveInstances(cluster); Assert.assertTrue(PollUtils.poll(() -> zkclient.getChildren(instancesPath).size() == 2, 100, 5000)); assignmentPath = KeyBuilder.instanceAssignments(cluster, instance1); Assert.assertTrue(PollUtils.poll((path) -> zkclient.getChildren(path).size() == 1, 100, 10000, assignmentPath)); LOG.info("Instance1 got task: " + zkclient.getChildren(assignmentPath)); String instance2 = server2.getCoordinator().getInstanceName(); assignmentPath = KeyBuilder.instanceAssignments(cluster, instance2); Assert.assertTrue(PollUtils.poll((path) -> zkclient.getChildren(path).size() == 1, 100, 10000, assignmentPath)); LOG.info("Instance2 got task: " + zkclient.getChildren(assignmentPath)); Thread.sleep(3000); eventsWritten1 = TestUtils.generateStrings(totalEvents); eventsWritten2 = TestUtils.generateStrings(totalEvents); FileUtils.writeLines(new File(fileName1), eventsWritten1, true ); FileUtils.writeLines(new File(fileName2), eventsWritten2, true ); eventsReceived1 = readFileDatastreamEvents(fileDatastream1, totalEvents * 2); eventsReceived2 = readFileDatastreamEvents(fileDatastream2, totalEvents * 2); LOG.info("(1-NEW) Events Received " + eventsReceived1); LOG.info("(1-NEW) Events Written to file " + eventsWritten1); LOG.info("(2-NEW) Events Received " + eventsReceived2); LOG.info("(2-NEW) Events Written to file " + eventsWritten2); Assert.assertTrue(eventsReceived1.containsAll(eventsWritten1)); Assert.assertTrue(eventsReceived2.containsAll(eventsWritten2)); }
public Coordinator getCoordinator() { return _coordinator; }
DatastreamServer { public Coordinator getCoordinator() { return _coordinator; } }
DatastreamServer { public Coordinator getCoordinator() { return _coordinator; } DatastreamServer(Properties properties); }
DatastreamServer { public Coordinator getCoordinator() { return _coordinator; } DatastreamServer(Properties properties); synchronized boolean isInitialized(); boolean isStarted(); Coordinator getCoordinator(); List<BrooklinMetricInfo> getMetricInfos(); DatastreamStore getDatastreamStore(); int getHttpPort(); ServerComponentHealthAggregator getServerComponentHealthAggregator(); synchronized void startup(); synchronized void shutdown(); static void main(String[] args); static Properties loadProps(String filename); }
DatastreamServer { public Coordinator getCoordinator() { return _coordinator; } DatastreamServer(Properties properties); synchronized boolean isInitialized(); boolean isStarted(); Coordinator getCoordinator(); List<BrooklinMetricInfo> getMetricInfos(); DatastreamStore getDatastreamStore(); int getHttpPort(); ServerComponentHealthAggregator getServerComponentHealthAggregator(); synchronized void startup(); synchronized void shutdown(); static void main(String[] args); static Properties loadProps(String filename);  }