@SuppressWarnings("unchecked") @Test public void testGetRecordList() throws Exception { SslContext sslCtx = ClientSSL.createInsecureContext(); 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, 0, records).get(); client.setLowWaterMark(SESSION_ID, 0, numRecords - 1).get(); client.close(); ArrayList<Record> oneRecordList = (ArrayList<Record>) adminClient.getRecordList(0, 0, 1).get(); ArrayList<Record> nineRecordList = (ArrayList<Record>) adminClient.getRecordList(0, 1, numRecords - 1).get(); adminClient.close(); assertEquals(1, oneRecordList.size()); assertEquals(records.get(0), oneRecordList.get(0)); assertEquals(9, nineRecordList.size()); for (int i = 1; i < records.size(); ++i) { assertEquals(records.get(i), nineRecordList.get(i - 1)); } } finally { storageRunner.stop(); } }
public CompletableFuture<Object> getRecordList(int partitionId, long transactionId, int maxNumRecords) { return call(new RecordListRequest(seqNum.getAndIncrement(), partitionId, transactionId, maxNumRecords)); }
StorageAdminClient extends StorageBaseClient { public CompletableFuture<Object> getRecordList(int partitionId, long transactionId, int maxNumRecords) { return call(new RecordListRequest(seqNum.getAndIncrement(), partitionId, transactionId, maxNumRecords)); } }
StorageAdminClient extends StorageBaseClient { public CompletableFuture<Object> getRecordList(int partitionId, long transactionId, int maxNumRecords) { return call(new RecordListRequest(seqNum.getAndIncrement(), partitionId, transactionId, maxNumRecords)); } StorageAdminClient(String host, int port, SslContext sslCtx, UUID key, int numPartitions); }
StorageAdminClient extends StorageBaseClient { public CompletableFuture<Object> getRecordList(int partitionId, long transactionId, int maxNumRecords) { return call(new RecordListRequest(seqNum.getAndIncrement(), partitionId, transactionId, maxNumRecords)); } 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> getRecordList(int partitionId, long transactionId, int maxNumRecords) { return call(new RecordListRequest(seqNum.getAndIncrement(), partitionId, transactionId, maxNumRecords)); } 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);  }