@SuppressWarnings("unchecked") @Test public void testRecoveryAlreadyDone() throws Exception { int partitionId = 0; long maxTransactionId = 35L; long lowWaterMark = 32L; long localLowWaterMark = 31L; long sessionId = 123L; SessionInfo sessionInfo = new SessionInfo(sessionId, lowWaterMark, localLowWaterMark); CompletableFuture<Object> sessionInfoCompletableFuture = mock(CompletableFuture.class); CompletableFuture<Object> maxTransactionIdCompletableFuture = mock(CompletableFuture.class); CompletableFuture<Object> lowWatermarkCompletableFuture = mock(CompletableFuture.class); StorageAdminClient mockSourceStorageAdminClient = mock(StorageAdminClient.class); StorageAdminClient mockDestinationStorageAdminClient = mock(StorageAdminClient.class); StorageClient mockDestinationStorageClient = mock(StorageClient.class); when(sessionInfoCompletableFuture.get()).thenReturn(sessionInfo); when(maxTransactionIdCompletableFuture.get()).thenReturn(maxTransactionId); when(lowWatermarkCompletableFuture.get()).thenReturn(lowWaterMark); when(mockSourceStorageAdminClient.lastSessionInfo(partitionId)).thenReturn(sessionInfoCompletableFuture); when(mockDestinationStorageAdminClient.lastSessionInfo(partitionId)).thenReturn(sessionInfoCompletableFuture); when(mockDestinationStorageClient.lastSessionInfo(sessionId, partitionId)).thenReturn(sessionInfoCompletableFuture); when(mockDestinationStorageClient.getMaxTransactionId(sessionId, partitionId)).thenReturn(maxTransactionIdCompletableFuture); when(mockDestinationStorageClient.truncate(sessionId, partitionId, lowWaterMark)).thenReturn(lowWatermarkCompletableFuture); StorageRecoveryRunnable storageRecoveryRunnable = new StorageRecoveryRunnable(mockSourceStorageAdminClient, mockDestinationStorageAdminClient, mockDestinationStorageClient, partitionId); storageRecoveryRunnable.run(); verify(mockSourceStorageAdminClient, never()).getRecordList(anyInt(), anyLong(), anyInt()); verify(mockSourceStorageAdminClient, times(1)).lastSessionInfo(partitionId); }
@SuppressWarnings("unchecked") @Override public void run() { ReplicaConnection destinationReplicaConnection = null; try { SessionInfo sessionInfo = (SessionInfo) destinationStorageAdminClient.lastSessionInfo(partitionId).get(); long destinationSessionId = sessionInfo.sessionId; long destinationLowWatermark = sessionInfo.lowWaterMark; destinationReplicaConnection = new ReplicaConnectionImpl(partitionId, destinationSessionId, destinationStorageClient); long currentHighWaterMark = destinationReplicaConnection.getMaxTransactionId(); logger.debug("destination high watermark = " + currentHighWaterMark); currentHighWaterMark = destinationReplicaConnection.truncate(destinationLowWatermark); logger.info("destination truncated to low watermark = " + destinationLowWatermark); SessionInfo sourceSessionInfo = (SessionInfo) sourceStorageAdminClient.lastSessionInfo(destinationReplicaConnection.getPartitionId()).get(); long targetLowWaterMark = sourceSessionInfo.localLowWaterMark; logger.info("source target low watermark = " + targetLowWaterMark); while (currentHighWaterMark < targetLowWaterMark) { int batchSize = (int) Math.min((targetLowWaterMark - currentHighWaterMark), (long) maxBatchSize); ArrayList<Record> records = (ArrayList<Record>) sourceStorageAdminClient .getRecordList(destinationReplicaConnection.getPartitionId(), currentHighWaterMark + 1, batchSize) .get(); destinationReplicaConnection.appendRecords(records); currentHighWaterMark += records.size(); destinationReplicaConnection.setLowWaterMark(currentHighWaterMark); logger.debug("destination high watermark = " + currentHighWaterMark); if (currentHighWaterMark % LOG_SIZE == 0) { float percentProgress = currentHighWaterMark / (float) targetLowWaterMark; logger.info(String.format("destination high watermark = %d (%.2f%%)", currentHighWaterMark, percentProgress)); } } } catch (StorageRpcException | InterruptedException | ExecutionException e) { throw new RuntimeException(e); } finally { if (destinationReplicaConnection != null) { destinationReplicaConnection.close(); } } }
StorageRecoveryRunnable implements Runnable { @SuppressWarnings("unchecked") @Override public void run() { ReplicaConnection destinationReplicaConnection = null; try { SessionInfo sessionInfo = (SessionInfo) destinationStorageAdminClient.lastSessionInfo(partitionId).get(); long destinationSessionId = sessionInfo.sessionId; long destinationLowWatermark = sessionInfo.lowWaterMark; destinationReplicaConnection = new ReplicaConnectionImpl(partitionId, destinationSessionId, destinationStorageClient); long currentHighWaterMark = destinationReplicaConnection.getMaxTransactionId(); logger.debug("destination high watermark = " + currentHighWaterMark); currentHighWaterMark = destinationReplicaConnection.truncate(destinationLowWatermark); logger.info("destination truncated to low watermark = " + destinationLowWatermark); SessionInfo sourceSessionInfo = (SessionInfo) sourceStorageAdminClient.lastSessionInfo(destinationReplicaConnection.getPartitionId()).get(); long targetLowWaterMark = sourceSessionInfo.localLowWaterMark; logger.info("source target low watermark = " + targetLowWaterMark); while (currentHighWaterMark < targetLowWaterMark) { int batchSize = (int) Math.min((targetLowWaterMark - currentHighWaterMark), (long) maxBatchSize); ArrayList<Record> records = (ArrayList<Record>) sourceStorageAdminClient .getRecordList(destinationReplicaConnection.getPartitionId(), currentHighWaterMark + 1, batchSize) .get(); destinationReplicaConnection.appendRecords(records); currentHighWaterMark += records.size(); destinationReplicaConnection.setLowWaterMark(currentHighWaterMark); logger.debug("destination high watermark = " + currentHighWaterMark); if (currentHighWaterMark % LOG_SIZE == 0) { float percentProgress = currentHighWaterMark / (float) targetLowWaterMark; logger.info(String.format("destination high watermark = %d (%.2f%%)", currentHighWaterMark, percentProgress)); } } } catch (StorageRpcException | InterruptedException | ExecutionException e) { throw new RuntimeException(e); } finally { if (destinationReplicaConnection != null) { destinationReplicaConnection.close(); } } } }
StorageRecoveryRunnable implements Runnable { @SuppressWarnings("unchecked") @Override public void run() { ReplicaConnection destinationReplicaConnection = null; try { SessionInfo sessionInfo = (SessionInfo) destinationStorageAdminClient.lastSessionInfo(partitionId).get(); long destinationSessionId = sessionInfo.sessionId; long destinationLowWatermark = sessionInfo.lowWaterMark; destinationReplicaConnection = new ReplicaConnectionImpl(partitionId, destinationSessionId, destinationStorageClient); long currentHighWaterMark = destinationReplicaConnection.getMaxTransactionId(); logger.debug("destination high watermark = " + currentHighWaterMark); currentHighWaterMark = destinationReplicaConnection.truncate(destinationLowWatermark); logger.info("destination truncated to low watermark = " + destinationLowWatermark); SessionInfo sourceSessionInfo = (SessionInfo) sourceStorageAdminClient.lastSessionInfo(destinationReplicaConnection.getPartitionId()).get(); long targetLowWaterMark = sourceSessionInfo.localLowWaterMark; logger.info("source target low watermark = " + targetLowWaterMark); while (currentHighWaterMark < targetLowWaterMark) { int batchSize = (int) Math.min((targetLowWaterMark - currentHighWaterMark), (long) maxBatchSize); ArrayList<Record> records = (ArrayList<Record>) sourceStorageAdminClient .getRecordList(destinationReplicaConnection.getPartitionId(), currentHighWaterMark + 1, batchSize) .get(); destinationReplicaConnection.appendRecords(records); currentHighWaterMark += records.size(); destinationReplicaConnection.setLowWaterMark(currentHighWaterMark); logger.debug("destination high watermark = " + currentHighWaterMark); if (currentHighWaterMark % LOG_SIZE == 0) { float percentProgress = currentHighWaterMark / (float) targetLowWaterMark; logger.info(String.format("destination high watermark = %d (%.2f%%)", currentHighWaterMark, percentProgress)); } } } catch (StorageRpcException | InterruptedException | ExecutionException e) { throw new RuntimeException(e); } finally { if (destinationReplicaConnection != null) { destinationReplicaConnection.close(); } } } StorageRecoveryRunnable(StorageAdminClient sourceStorageAdminClient, StorageAdminClient destinationStorageAdminClient, StorageClient destinationStorageClient, int partitionId);  StorageRecoveryRunnable(StorageAdminClient sourceStorageAdminClient, StorageAdminClient destinationStorageAdminClient, StorageClient destinationStorageClient, int partitionId, int maxBatchSize); }
StorageRecoveryRunnable implements Runnable { @SuppressWarnings("unchecked") @Override public void run() { ReplicaConnection destinationReplicaConnection = null; try { SessionInfo sessionInfo = (SessionInfo) destinationStorageAdminClient.lastSessionInfo(partitionId).get(); long destinationSessionId = sessionInfo.sessionId; long destinationLowWatermark = sessionInfo.lowWaterMark; destinationReplicaConnection = new ReplicaConnectionImpl(partitionId, destinationSessionId, destinationStorageClient); long currentHighWaterMark = destinationReplicaConnection.getMaxTransactionId(); logger.debug("destination high watermark = " + currentHighWaterMark); currentHighWaterMark = destinationReplicaConnection.truncate(destinationLowWatermark); logger.info("destination truncated to low watermark = " + destinationLowWatermark); SessionInfo sourceSessionInfo = (SessionInfo) sourceStorageAdminClient.lastSessionInfo(destinationReplicaConnection.getPartitionId()).get(); long targetLowWaterMark = sourceSessionInfo.localLowWaterMark; logger.info("source target low watermark = " + targetLowWaterMark); while (currentHighWaterMark < targetLowWaterMark) { int batchSize = (int) Math.min((targetLowWaterMark - currentHighWaterMark), (long) maxBatchSize); ArrayList<Record> records = (ArrayList<Record>) sourceStorageAdminClient .getRecordList(destinationReplicaConnection.getPartitionId(), currentHighWaterMark + 1, batchSize) .get(); destinationReplicaConnection.appendRecords(records); currentHighWaterMark += records.size(); destinationReplicaConnection.setLowWaterMark(currentHighWaterMark); logger.debug("destination high watermark = " + currentHighWaterMark); if (currentHighWaterMark % LOG_SIZE == 0) { float percentProgress = currentHighWaterMark / (float) targetLowWaterMark; logger.info(String.format("destination high watermark = %d (%.2f%%)", currentHighWaterMark, percentProgress)); } } } catch (StorageRpcException | InterruptedException | ExecutionException e) { throw new RuntimeException(e); } finally { if (destinationReplicaConnection != null) { destinationReplicaConnection.close(); } } } StorageRecoveryRunnable(StorageAdminClient sourceStorageAdminClient, StorageAdminClient destinationStorageAdminClient, StorageClient destinationStorageClient, int partitionId);  StorageRecoveryRunnable(StorageAdminClient sourceStorageAdminClient, StorageAdminClient destinationStorageAdminClient, StorageClient destinationStorageClient, int partitionId, int maxBatchSize); @SuppressWarnings("unchecked") @Override void run(); }
StorageRecoveryRunnable implements Runnable { @SuppressWarnings("unchecked") @Override public void run() { ReplicaConnection destinationReplicaConnection = null; try { SessionInfo sessionInfo = (SessionInfo) destinationStorageAdminClient.lastSessionInfo(partitionId).get(); long destinationSessionId = sessionInfo.sessionId; long destinationLowWatermark = sessionInfo.lowWaterMark; destinationReplicaConnection = new ReplicaConnectionImpl(partitionId, destinationSessionId, destinationStorageClient); long currentHighWaterMark = destinationReplicaConnection.getMaxTransactionId(); logger.debug("destination high watermark = " + currentHighWaterMark); currentHighWaterMark = destinationReplicaConnection.truncate(destinationLowWatermark); logger.info("destination truncated to low watermark = " + destinationLowWatermark); SessionInfo sourceSessionInfo = (SessionInfo) sourceStorageAdminClient.lastSessionInfo(destinationReplicaConnection.getPartitionId()).get(); long targetLowWaterMark = sourceSessionInfo.localLowWaterMark; logger.info("source target low watermark = " + targetLowWaterMark); while (currentHighWaterMark < targetLowWaterMark) { int batchSize = (int) Math.min((targetLowWaterMark - currentHighWaterMark), (long) maxBatchSize); ArrayList<Record> records = (ArrayList<Record>) sourceStorageAdminClient .getRecordList(destinationReplicaConnection.getPartitionId(), currentHighWaterMark + 1, batchSize) .get(); destinationReplicaConnection.appendRecords(records); currentHighWaterMark += records.size(); destinationReplicaConnection.setLowWaterMark(currentHighWaterMark); logger.debug("destination high watermark = " + currentHighWaterMark); if (currentHighWaterMark % LOG_SIZE == 0) { float percentProgress = currentHighWaterMark / (float) targetLowWaterMark; logger.info(String.format("destination high watermark = %d (%.2f%%)", currentHighWaterMark, percentProgress)); } } } catch (StorageRpcException | InterruptedException | ExecutionException e) { throw new RuntimeException(e); } finally { if (destinationReplicaConnection != null) { destinationReplicaConnection.close(); } } } StorageRecoveryRunnable(StorageAdminClient sourceStorageAdminClient, StorageAdminClient destinationStorageAdminClient, StorageClient destinationStorageClient, int partitionId);  StorageRecoveryRunnable(StorageAdminClient sourceStorageAdminClient, StorageAdminClient destinationStorageAdminClient, StorageClient destinationStorageClient, int partitionId, int maxBatchSize); @SuppressWarnings("unchecked") @Override void run();  }