|
@@ -403,17 +403,25 @@ public class RecoverySourceHandler {
|
|
|
logger.trace("recovery [phase1]: recovering_files [{}] with total_size [{}], reusing_files [{}] with total_size [{}]",
|
|
|
phase1FileNames.size(), new ByteSizeValue(totalSizeInBytes),
|
|
|
phase1ExistingFileNames.size(), new ByteSizeValue(existingTotalSizeInBytes));
|
|
|
- cancellableThreads.execute(() -> recoveryTarget.receiveFileInfo(
|
|
|
- phase1FileNames, phase1FileSizes, phase1ExistingFileNames, phase1ExistingFileSizes, translogOps.getAsInt()));
|
|
|
- sendFiles(store, phase1Files.toArray(new StoreFileMetaData[0]), translogOps);
|
|
|
+ final StepListener<Void> sendFileInfoStep = new StepListener<>();
|
|
|
+ final StepListener<Void> cleanFilesStep = new StepListener<>();
|
|
|
+ cancellableThreads.execute(() ->
|
|
|
+ recoveryTarget.receiveFileInfo(phase1FileNames, phase1FileSizes, phase1ExistingFileNames,
|
|
|
+ phase1ExistingFileSizes, translogOps.getAsInt(), sendFileInfoStep));
|
|
|
+
|
|
|
+ sendFileInfoStep.whenComplete(r -> {
|
|
|
+ sendFiles(store, phase1Files.toArray(new StoreFileMetaData[0]), translogOps);
|
|
|
+ cleanFiles(store, recoverySourceMetadata, translogOps, globalCheckpoint, cleanFilesStep);
|
|
|
+ }, listener::onFailure);
|
|
|
+
|
|
|
final long totalSize = totalSizeInBytes;
|
|
|
final long existingTotalSize = existingTotalSizeInBytes;
|
|
|
- cleanFiles(store, recoverySourceMetadata, translogOps, globalCheckpoint, ActionListener.map(listener, aVoid -> {
|
|
|
+ cleanFilesStep.whenComplete(r -> {
|
|
|
final TimeValue took = stopWatch.totalTime();
|
|
|
logger.trace("recovery [phase1]: took [{}]", took);
|
|
|
- return new SendFileResult(phase1FileNames, phase1FileSizes, totalSize, phase1ExistingFileNames,
|
|
|
- phase1ExistingFileSizes, existingTotalSize, took);
|
|
|
- }));
|
|
|
+ listener.onResponse(new SendFileResult(phase1FileNames, phase1FileSizes, totalSize, phase1ExistingFileNames,
|
|
|
+ phase1ExistingFileSizes, existingTotalSize, took));
|
|
|
+ }, listener::onFailure);
|
|
|
} else {
|
|
|
logger.trace("skipping [phase1]- identical sync id [{}] found on both source and target",
|
|
|
recoverySourceMetadata.getSyncId());
|