@Test public void testSetLowWaterMark() throws Exception { 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(); } int port = waltzStorage.port; StorageClient client = new StorageClient(host, port, sslCtx, key, NUM_PARTITIONS); client.open(); int transactionId; ArrayList<Record> records; SessionInfo sessionInfo = (SessionInfo) client.lastSessionInfo(SESSION_ID, 0).get(); assertEquals(-1L, sessionInfo.sessionId); assertEquals(-1L, sessionInfo.lowWaterMark); client.setLowWaterMark(SESSION_ID, 0, -1L).get(); transactionId = 0; records = ClientUtil.makeRecords(transactionId, transactionId + 10); transactionId += records.size(); client.appendRecords(SESSION_ID, 0, records); assertEquals((long) (records.size() - 1), client.getMaxTransactionId(SESSION_ID, 0).get()); sessionInfo = (SessionInfo) client.lastSessionInfo(SESSION_ID, 0).get(); assertEquals(0L, sessionInfo.sessionId); assertEquals(-1L, sessionInfo.lowWaterMark); sessionInfo = (SessionInfo) client.lastSessionInfo(SESSION_ID + 1, 0).get(); assertEquals(0L, sessionInfo.sessionId); assertEquals(-1L, sessionInfo.lowWaterMark); client.setLowWaterMark(SESSION_ID + 1, 0, transactionId + 10); sessionInfo = (SessionInfo) client.lastSessionInfo(SESSION_ID + 1, 0).get(); assertEquals(1L, sessionInfo.sessionId); assertEquals(transactionId + 10, sessionInfo.lowWaterMark); client.close(); } finally { storageRunner.stop(); } }
public CompletableFuture<Object> setLowWaterMark(long sessionId, int partitionId, long lowWaterMark) { return call(new SetLowWaterMarkRequest(sessionId, seqNum.getAndIncrement(), partitionId, lowWaterMark, usedByOfflineRecovery)); }
StorageClient extends StorageBaseClient { public CompletableFuture<Object> setLowWaterMark(long sessionId, int partitionId, long lowWaterMark) { return call(new SetLowWaterMarkRequest(sessionId, seqNum.getAndIncrement(), partitionId, lowWaterMark, usedByOfflineRecovery)); } }
StorageClient extends StorageBaseClient { public CompletableFuture<Object> setLowWaterMark(long sessionId, int partitionId, long lowWaterMark) { return call(new SetLowWaterMarkRequest(sessionId, seqNum.getAndIncrement(), partitionId, lowWaterMark, usedByOfflineRecovery)); } StorageClient(String host, int port, SslContext sslCtx, UUID key, int numPartitions);  StorageClient(String host, int port, SslContext sslCtx, UUID key, int numPartitions, boolean usedByOfflineRecovery); }
StorageClient extends StorageBaseClient { public CompletableFuture<Object> setLowWaterMark(long sessionId, int partitionId, long lowWaterMark) { return call(new SetLowWaterMarkRequest(sessionId, seqNum.getAndIncrement(), partitionId, lowWaterMark, usedByOfflineRecovery)); } StorageClient(String host, int port, SslContext sslCtx, UUID key, int numPartitions);  StorageClient(String host, int port, SslContext sslCtx, UUID key, int numPartitions, boolean usedByOfflineRecovery); CompletableFuture<Object> lastSessionInfo(long sessionId, int partitionId); CompletableFuture<Object> setLowWaterMark(long sessionId, int partitionId, long lowWaterMark); CompletableFuture<Object> truncate(long sessionId, int partitionId, long transactionId); CompletableFuture<Object> getRecordHeader(long sessionId, int partitionId, long transactionId); CompletableFuture<Object> getRecordHeaderList(long sessionId, int partitionId, long transactionId, int maxNumRecords); CompletableFuture<Object> getRecord(long sessionId, int partitionId, long transactionId); CompletableFuture<Object> getRecordList(long sessionId, int partitionId, long transactionId, int maxNumRecords); CompletableFuture<Object> getMaxTransactionId(long sessionId, int partitionId); CompletableFuture<Object> appendRecords(long sessionId, int partitionId, ArrayList<Record> records); }
StorageClient extends StorageBaseClient { public CompletableFuture<Object> setLowWaterMark(long sessionId, int partitionId, long lowWaterMark) { return call(new SetLowWaterMarkRequest(sessionId, seqNum.getAndIncrement(), partitionId, lowWaterMark, usedByOfflineRecovery)); } StorageClient(String host, int port, SslContext sslCtx, UUID key, int numPartitions);  StorageClient(String host, int port, SslContext sslCtx, UUID key, int numPartitions, boolean usedByOfflineRecovery); CompletableFuture<Object> lastSessionInfo(long sessionId, int partitionId); CompletableFuture<Object> setLowWaterMark(long sessionId, int partitionId, long lowWaterMark); CompletableFuture<Object> truncate(long sessionId, int partitionId, long transactionId); CompletableFuture<Object> getRecordHeader(long sessionId, int partitionId, long transactionId); CompletableFuture<Object> getRecordHeaderList(long sessionId, int partitionId, long transactionId, int maxNumRecords); CompletableFuture<Object> getRecord(long sessionId, int partitionId, long transactionId); CompletableFuture<Object> getRecordList(long sessionId, int partitionId, long transactionId, int maxNumRecords); CompletableFuture<Object> getMaxTransactionId(long sessionId, int partitionId); CompletableFuture<Object> appendRecords(long sessionId, int partitionId, ArrayList<Record> records);  }