### Question:
ReplicaConnectionImpl implements ReplicaConnection { public void setLowWaterMark(long lowWaterMark) throws StorageRpcException { get(client.setLowWaterMark(sessionId, partitionId, lowWaterMark)); } 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();  }### Answer:
@Test public void testSetLowWaterMark() throws Exception { int transactionId = 0; try (ReplicaConnection replicaConnection0 = connectionFactory.get(0, 0)) { SessionInfo sessionInfo = replicaConnection0.lastSessionInfo(); assertEquals(-1L, sessionInfo.sessionId); assertEquals(-1L, sessionInfo.lowWaterMark); replicaConnection0.setLowWaterMark(-1L); ArrayList<Record> records = makeRecords(transactionId, transactionId + 10); transactionId += records.size(); replicaConnection0.appendRecords(records); assertEquals(records.size() - 1, replicaConnection0.getMaxTransactionId()); } try (ReplicaConnection replicaConnection1 = connectionFactory.get(0, 1)) { SessionInfo sessionInfo = replicaConnection1.lastSessionInfo(); assertEquals(0L, sessionInfo.sessionId); assertEquals(-1L, sessionInfo.lowWaterMark); replicaConnection1.setLowWaterMark(transactionId + 10); } try (ReplicaConnection replicaConnection2 = connectionFactory.get(0, 2)) { SessionInfo sessionInfo = replicaConnection2.lastSessionInfo(); assertEquals(1L, sessionInfo.sessionId); assertEquals(transactionId + 10, sessionInfo.lowWaterMark); } }