@Test public void testUserManagedDestination() throws Exception { int numberOfPartitions = 2; String destinationTopic = "userManaged_" + UUID.randomUUID().toString(); _datastreamCluster = initializeTestDatastreamServerWithFileConnector(1, BROADCAST_STRATEGY_FACTORY, numberOfPartitions); int totalEvents = 10; _datastreamCluster.startup(); Path tempFile1 = Files.createTempFile("testFile1", ""); String fileName1 = tempFile1.toAbsolutePath().toString(); ZkClient zkClient = new ZkClient(_datastreamCluster.getZkConnection()); ZkConnection zkConnection = new ZkConnection(_datastreamCluster.getZkConnection()); ZkUtils zkUtils = new ZkUtils(zkClient, zkConnection, false); AdminUtils.createTopic(zkUtils, destinationTopic, numberOfPartitions, 1, new Properties(), RackAwareMode.Disabled$.MODULE$); Datastream fileDatastream1 = createFileDatastream(fileName1, destinationTopic, 2); Assert.assertEquals((int) fileDatastream1.getDestination().getPartitions(), 2); Collection<String> eventsWritten1 = TestUtils.generateStrings(totalEvents); FileUtils.writeLines(new File(fileName1), eventsWritten1); Collection<String> eventsReceived1 = readFileDatastreamEvents(fileDatastream1, 0, 4); Collection<String> eventsReceived2 = readFileDatastreamEvents(fileDatastream1, 1, 6); eventsReceived1.addAll(eventsReceived2); LOG.info("Events Received " + eventsReceived1); LOG.info("Events Written to file " + eventsWritten1); Assert.assertTrue(eventsReceived1.containsAll(eventsWritten1)); }
public synchronized void startup() throws DatastreamException { if (_jmxReporter != null) { _jmxReporter.start(); } if (_coordinator != null) { _coordinator.start(); } try { _jettyLauncher.start(); _httpPort = _jettyLauncher.getPort(); _serverComponentHealthAggregator.setPort(_httpPort); _isStarted = true; } catch (Exception ex) { ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, "Failed to start embedded Jetty.", ex); } }
DatastreamServer { public synchronized void startup() throws DatastreamException { if (_jmxReporter != null) { _jmxReporter.start(); } if (_coordinator != null) { _coordinator.start(); } try { _jettyLauncher.start(); _httpPort = _jettyLauncher.getPort(); _serverComponentHealthAggregator.setPort(_httpPort); _isStarted = true; } catch (Exception ex) { ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, "Failed to start embedded Jetty.", ex); } } }
DatastreamServer { public synchronized void startup() throws DatastreamException { if (_jmxReporter != null) { _jmxReporter.start(); } if (_coordinator != null) { _coordinator.start(); } try { _jettyLauncher.start(); _httpPort = _jettyLauncher.getPort(); _serverComponentHealthAggregator.setPort(_httpPort); _isStarted = true; } catch (Exception ex) { ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, "Failed to start embedded Jetty.", ex); } } DatastreamServer(Properties properties); }
DatastreamServer { public synchronized void startup() throws DatastreamException { if (_jmxReporter != null) { _jmxReporter.start(); } if (_coordinator != null) { _coordinator.start(); } try { _jettyLauncher.start(); _httpPort = _jettyLauncher.getPort(); _serverComponentHealthAggregator.setPort(_httpPort); _isStarted = true; } catch (Exception ex) { ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, "Failed to start embedded Jetty.", ex); } } 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 synchronized void startup() throws DatastreamException { if (_jmxReporter != null) { _jmxReporter.start(); } if (_coordinator != null) { _coordinator.start(); } try { _jettyLauncher.start(); _httpPort = _jettyLauncher.getPort(); _serverComponentHealthAggregator.setPort(_httpPort); _isStarted = true; } catch (Exception ex) { ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, "Failed to start embedded Jetty.", ex); } } 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);  }