@Test public void testTruncate() throws Exception { final ArrayList<Record> records = makeRecords(0, 20); final int transactionId = records.size(); try (ReplicaConnection replicaConnection0 = connectionFactory.get(0, 0)) { replicaConnection0.lastSessionInfo(); replicaConnection0.setLowWaterMark(-1L); replicaConnection0.appendRecords(records); assertEquals(records.size() - 1, replicaConnection0.getMaxTransactionId()); } try (ReplicaConnection replicaConnection1 = connectionFactory.get(0, 1)) { replicaConnection1.lastSessionInfo(); long newMaxAfterTruncate = transactionId / 2; assertEquals(newMaxAfterTruncate, replicaConnection1.truncate(newMaxAfterTruncate)); replicaConnection1.setLowWaterMark(newMaxAfterTruncate); assertEquals(newMaxAfterTruncate, replicaConnection1.getMaxTransactionId()); for (Record record : records) { Record returnedRecord = replicaConnection1.getRecord(record.transactionId); if (record.transactionId <= newMaxAfterTruncate) { assertNotNull(returnedRecord); assertEquals(record.transactionId, returnedRecord.transactionId); assertEquals(record.reqId, returnedRecord.reqId); assertEquals(record.header, returnedRecord.header); assertArrayEquals(record.data, returnedRecord.data); assertEquals(record.checksum, returnedRecord.checksum); } else { assertNull(returnedRecord); } } } }
public long truncate(long transactionId) throws StorageRpcException { return (Long) get(client.truncate(sessionId, partitionId, transactionId)); }
ReplicaConnectionImpl implements ReplicaConnection { public long truncate(long transactionId) throws StorageRpcException { return (Long) get(client.truncate(sessionId, partitionId, transactionId)); } }
ReplicaConnectionImpl implements ReplicaConnection { public long truncate(long transactionId) throws StorageRpcException { return (Long) get(client.truncate(sessionId, partitionId, transactionId)); } ReplicaConnectionImpl(int partitionId, long sessionId, StorageClient client); }
ReplicaConnectionImpl implements ReplicaConnection { public long truncate(long transactionId) throws StorageRpcException { return (Long) get(client.truncate(sessionId, partitionId, transactionId)); } ReplicaConnectionImpl(int partitionId, long sessionId, StorageClient client); SessionInfo lastSessionInfo(); void setLowWaterMark(long lowWaterMark); long truncate(long transactionId); RecordHeader getRecordHeader(long transactionId); @SuppressWarnings("unchecked") ArrayList<RecordHeader> getRecordHeaderList(long transactionId, int maxNumRecords); Record getRecord(long transactionId); @SuppressWarnings("unchecked") ArrayList<Record> getRecordList(long transactionId, int maxNumRecords); long getMaxTransactionId(); @Override int getPartitionId(); void appendRecords(ArrayList<Record> records); void close(); }
ReplicaConnectionImpl implements ReplicaConnection { public long truncate(long transactionId) throws StorageRpcException { return (Long) get(client.truncate(sessionId, partitionId, transactionId)); } ReplicaConnectionImpl(int partitionId, long sessionId, StorageClient client); SessionInfo lastSessionInfo(); void setLowWaterMark(long lowWaterMark); long truncate(long transactionId); RecordHeader getRecordHeader(long transactionId); @SuppressWarnings("unchecked") ArrayList<RecordHeader> getRecordHeaderList(long transactionId, int maxNumRecords); Record getRecord(long transactionId); @SuppressWarnings("unchecked") ArrayList<Record> getRecordList(long transactionId, int maxNumRecords); long getMaxTransactionId(); @Override int getPartitionId(); void appendRecords(ArrayList<Record> records); void close();  }