DistributedLogServer { public void close() { if (null != announcer) { try { announcer.unannounce(); } catch (IOException e) { logger.warn("Error on unannouncing service : ", e); } announcer.close(); } closeServer(Pair.of(dlService, server), gracefulShutdownMs, TimeUnit.MILLISECONDS); if (null != statsProvider) { statsProvider.stop(); } SchedulerUtils.shutdownScheduler(configExecutorService, 60, TimeUnit.SECONDS); keepAliveLatch.countDown(); } DistributedLogServer(Optional<String> uri,
                         Optional<String> conf,
                         Optional<String> streamConf,
                         Optional<Integer> port,
                         Optional<Integer> statsPort,
                         Optional<Integer> shardId,
                         Optional<Boolean> announceServerSet,
                         Optional<Boolean> thriftmux,
                         StatsReceiver statsReceiver,
                         StatsProvider statsProvider); void runServer(); void close(); void join(); static DistributedLogServer runServer(
               Optional<String> uri,
               Optional<String> conf,
               Optional<String> streamConf,
               Optional<Integer> port,
               Optional<Integer> statsPort,
               Optional<Integer> shardId,
               Optional<Boolean> announceServerSet,
               Optional<Boolean> thriftmux,
               StatsReceiver statsReceiver,
               StatsProvider statsProvider);  }
@DistributedLogAnnotations.FlakyTest @Ignore @Test(timeout = 60000) public void testBasicWrite() throws Exception { String name = "dlserver-basic-write"; dlClient.routingService.addHost(name, dlServer.getAddress()); for (long i = 1; i <= 10; i++) { logger.debug("Write entry {} to stream {}.", i, name); Await.result(dlClient.dlClient.write(name, ByteBuffer.wrap(("" + i).getBytes()))); } HeartbeatOptions hbOptions = new HeartbeatOptions(); hbOptions.setSendHeartBeatToReader(true); FutureUtils.result(dlClient.dlClient.heartbeat(name)); DistributedLogManager dlm = DLMTestUtil.createNewDLM(name, conf, getUri()); LogReader reader = dlm.getInputStream(1); int numRead = 0; LogRecord r = reader.readNext(false); while (null != r) { int i = Integer.parseInt(new String(r.getPayload())); assertEquals(numRead + 1, i); ++numRead; r = reader.readNext(false); } assertEquals(10, numRead); reader.close(); dlm.close(); }
@Test(timeout = 60000) public void testChecksumFlag() throws Exception { String name = "dlserver-basic-write"; LocalRoutingService routingService = LocalRoutingService.newBuilder().build(); routingService.addHost(name, dlServer.getAddress()); DistributedLogClientBuilder dlClientBuilder = DistributedLogClientBuilder.newBuilder() .name(name) .clientId(ClientId$.MODULE$.apply("test")) .routingService(routingService) .handshakeWithClientInfo(true) .clientBuilder(ClientBuilder.get() .hostConnectionLimit(1) .connectionTimeout(Duration.fromSeconds(1)) .requestTimeout(Duration.fromSeconds(60))) .checksum(false); DistributedLogClient dlClient = (DistributedLogClientImpl) dlClientBuilder.build(); Await.result(dlClient.write(name, ByteBuffer.wrap(("1").getBytes()))); dlClient.close(); dlClient = dlClientBuilder.checksum(true).build(); Await.result(dlClient.write(name, ByteBuffer.wrap(("2").getBytes()))); dlClient.close(); }
@Test(timeout = 60000) public void testFenceWrite() throws Exception { String name = "dlserver-fence-write"; dlClient.routingService.addHost(name, dlServer.getAddress()); for (long i = 1; i <= 10; i++) { logger.debug("Write entry {} to stream {}.", i, name); dlClient.dlClient.write(name, ByteBuffer.wrap(("" + i).getBytes())).get(); } Thread.sleep(1000); logger.info("Fencing stream {}.", name); DLMTestUtil.fenceStream(conf, getUri(), name); logger.info("Fenced stream {}.", name); for (long i = 11; i <= 20; i++) { logger.debug("Write entry {} to stream {}.", i, name); dlClient.dlClient.write(name, ByteBuffer.wrap(("" + i).getBytes())).get(); } DistributedLogManager dlm = DLMTestUtil.createNewDLM(name, conf, getUri()); LogReader reader = dlm.getInputStream(1); int numRead = 0; LogRecord r = reader.readNext(false); while (null != r) { int i = Integer.parseInt(new String(r.getPayload())); assertEquals(numRead + 1, i); ++numRead; r = reader.readNext(false); } assertEquals(20, numRead); reader.close(); dlm.close(); }
@Test(timeout = 60000) public void testDeleteStream() throws Exception { String name = "dlserver-delete-stream"; dlClient.routingService.addHost(name, dlServer.getAddress()); long txid = 101; for (long i = 1; i <= 10; i++) { long curTxId = txid++; logger.debug("Write entry {} to stream {}.", curTxId, name); dlClient.dlClient.write(name, ByteBuffer.wrap(("" + curTxId).getBytes())).get(); } checkStream(1, 1, 1, name, dlServer.getAddress(), true, true); dlClient.dlClient.delete(name).get(); checkStream(0, 0, 0, name, dlServer.getAddress(), false, false); Thread.sleep(1000); DistributedLogManager dlm101 = DLMTestUtil.createNewDLM(name, conf, getUri()); AsyncLogReader reader101 = FutureUtils.result(dlm101.openAsyncLogReader(DLSN.InitialDLSN)); try { FutureUtils.result(reader101.readNext()); fail("Should fail with LogNotFoundException since the stream is deleted"); } catch (LogNotFoundException lnfe) { } FutureUtils.result(reader101.asyncClose()); dlm101.close(); txid = 201; for (long i = 1; i <= 10; i++) { long curTxId = txid++; logger.debug("Write entry {} to stream {}.", curTxId, name); DLSN dlsn = dlClient.dlClient.write(name, ByteBuffer.wrap(("" + curTxId).getBytes())).get(); } Thread.sleep(1000); DistributedLogManager dlm201 = DLMTestUtil.createNewDLM(name, conf, getUri()); LogReader reader201 = dlm201.getInputStream(1); int numRead = 0; int curTxId = 201; LogRecord r = reader201.readNext(false); while (null != r) { int i = Integer.parseInt(new String(r.getPayload())); assertEquals(curTxId++, i); ++numRead; r = reader201.readNext(false); } assertEquals(10, numRead); reader201.close(); dlm201.close(); }
@Test(timeout = 60000) public void testTruncateStream() throws Exception { String name = "dlserver-truncate-stream"; dlClient.routingService.addHost(name, dlServer.getAddress()); long txid = 1; Map<Long, DLSN> txid2DLSN = new HashMap<Long, DLSN>(); for (int s = 1; s <= 3; s++) { for (long i = 1; i <= 10; i++) { long curTxId = txid++; logger.debug("Write entry {} to stream {}.", curTxId, name); DLSN dlsn = dlClient.dlClient.write(name, ByteBuffer.wrap(("" + curTxId).getBytes())).get(); txid2DLSN.put(curTxId, dlsn); } if (s <= 2) { dlClient.dlClient.release(name).get(); } } DLSN dlsnToDelete = txid2DLSN.get(21L); dlClient.dlClient.truncate(name, dlsnToDelete).get(); DistributedLogManager readDLM = DLMTestUtil.createNewDLM(name, conf, getUri()); LogReader reader = readDLM.getInputStream(1); int numRead = 0; int curTxId = 11; LogRecord r = reader.readNext(false); while (null != r) { int i = Integer.parseInt(new String(r.getPayload())); assertEquals(curTxId++, i); ++numRead; r = reader.readNext(false); } assertEquals(20, numRead); reader.close(); readDLM.close(); }