@SuppressWarnings("unchecked") @Test public void testLastSessionInfo() throws Exception { long newSessionId = SESSION_ID + 1L; WaltzStorageRunner storageRunner = new WaltzStorageRunner(portFinder, waltzStorageConfig, segmentSizeThreshold); try { storageRunner.startAsync(); WaltzStorage waltzStorage = storageRunner.awaitStart(); int adminPort = waltzStorage.adminPort; StorageAdminClient adminClient = new StorageAdminClient(host, adminPort, sslCtx, key, NUM_PARTITIONS); adminClient.open(); for (int i = 0; i < NUM_PARTITIONS; i++) { adminClient.setPartitionAssignment(i, true, false).get(); } ArrayList<Record> records = ClientUtil.makeRecords(0, 10); int numRecords = records.size(); int port = waltzStorage.port; StorageClient client = new StorageClient(host, port, sslCtx, key, NUM_PARTITIONS); client.open(); client.appendRecords(SESSION_ID, 1, records).get(); client.setLowWaterMark(newSessionId, 1, numRecords - 1).get(); client.close(); SessionInfo sessionInfo = (SessionInfo) adminClient.lastSessionInfo(1).get(); adminClient.close(); assertEquals(newSessionId, sessionInfo.sessionId); assertEquals(numRecords - 1, sessionInfo.lowWaterMark); } finally { storageRunner.stop(); } }
public CompletableFuture<Object> lastSessionInfo(int partitionId) { return call(new LastSessionInfoRequest(seqNum.getAndIncrement(), partitionId)); }
StorageAdminClient extends StorageBaseClient { public CompletableFuture<Object> lastSessionInfo(int partitionId) { return call(new LastSessionInfoRequest(seqNum.getAndIncrement(), partitionId)); } }
StorageAdminClient extends StorageBaseClient { public CompletableFuture<Object> lastSessionInfo(int partitionId) { return call(new LastSessionInfoRequest(seqNum.getAndIncrement(), partitionId)); } StorageAdminClient(String host, int port, SslContext sslCtx, UUID key, int numPartitions); }
StorageAdminClient extends StorageBaseClient { public CompletableFuture<Object> lastSessionInfo(int partitionId) { return call(new LastSessionInfoRequest(seqNum.getAndIncrement(), partitionId)); } StorageAdminClient(String host, int port, SslContext sslCtx, UUID key, int numPartitions); CompletableFuture<Object> setPartitionAvailable(int partitionId, boolean isAvailable); CompletableFuture<Object> setPartitionAssignment(int partitionId, boolean isAssigned, boolean deleteStorageFiles); CompletableFuture<Object> getMetrics(); CompletableFuture<Object> getRecordList(int partitionId, long transactionId, int maxNumRecords); CompletableFuture<Object> getAssignedPartitionStatus(); CompletableFuture<Object> lastSessionInfo(int partitionId); }
StorageAdminClient extends StorageBaseClient { public CompletableFuture<Object> lastSessionInfo(int partitionId) { return call(new LastSessionInfoRequest(seqNum.getAndIncrement(), partitionId)); } StorageAdminClient(String host, int port, SslContext sslCtx, UUID key, int numPartitions); CompletableFuture<Object> setPartitionAvailable(int partitionId, boolean isAvailable); CompletableFuture<Object> setPartitionAssignment(int partitionId, boolean isAssigned, boolean deleteStorageFiles); CompletableFuture<Object> getMetrics(); CompletableFuture<Object> getRecordList(int partitionId, long transactionId, int maxNumRecords); CompletableFuture<Object> getAssignedPartitionStatus(); CompletableFuture<Object> lastSessionInfo(int partitionId);  }