Skip to content

Commit

Permalink
Small cleanup in PRTS#doRecovery (#93549)
Browse files Browse the repository at this point in the history 
Renames `failureHandler` to `cleanupOnly` and makes use of
`ActionListener#run` where appropriate.
  • Loading branch information
@DaveCTurner
DaveCTurner committed Feb 7, 2023
1 parent b8c9dc9 commit 134f51b
Showing 1 changed file with 34 additions and 42 deletions.
76 changes: 34 additions & 42 deletions server/src/main/java/org/elasticsearch/indices/recovery/PeerRecoveryTargetService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -222,7 +222,8 @@ private void doRecovery(final long recoveryId, final StartRecoveryRequest preExi
final IndexShard indexShard = recoveryTarget.indexShard();
final Releasable onCompletion = Releasables.wrap(recoveryTarget.disableRecoveryMonitor(), recoveryRef);

final var failureHandler = ActionListener.notifyOnce(ActionListener.runBefore(ActionListener.noop().delegateResponse((l, e) -> {
// async version of the catch/finally structure we need, but this does nothing with successes so needs further modification below
final var cleanupOnly = ActionListener.notifyOnce(ActionListener.runBefore(ActionListener.noop().delegateResponse((l, e) -> {
// this will be logged as warning later on...
logger.trace("unexpected error while preparing shard for peer recovery, failing recovery", e);
onGoingRecoveries.failRecovery(
Expand All @@ -235,28 +236,23 @@ private void doRecovery(final long recoveryId, final StartRecoveryRequest preExi
if (indexShard.routingEntry().isPromotableToPrimary() == false) {
assert preExistingRequest == null;
assert indexShard.indexSettings().getIndexMetadata().isSearchableSnapshot() == false;
try {
indexShard.preRecovery(failureHandler.map(v -> {
logger.trace("{} preparing shard for peer recovery", recoveryTarget.shardId());
indexShard.prepareForIndexRecovery();
// Skip unnecessary intermediate stages
recoveryState.setStage(RecoveryState.Stage.VERIFY_INDEX);
recoveryState.setStage(RecoveryState.Stage.TRANSLOG);
indexShard.openEngineAndSkipTranslogRecovery();
recoveryState.getIndex().setFileDetailsComplete();
recoveryState.setStage(RecoveryState.Stage.FINALIZE);
onGoingRecoveries.markRecoveryAsDone(recoveryId);
return null;
}));
} catch (Exception e) {
failureHandler.onFailure(e);
}

ActionListener.run(cleanupOnly.map(v -> {
logger.trace("{} preparing shard for peer recovery", recoveryTarget.shardId());
indexShard.prepareForIndexRecovery();
// Skip unnecessary intermediate stages
recoveryState.setStage(RecoveryState.Stage.VERIFY_INDEX);
recoveryState.setStage(RecoveryState.Stage.TRANSLOG);
indexShard.openEngineAndSkipTranslogRecovery();
recoveryState.getIndex().setFileDetailsComplete();
recoveryState.setStage(RecoveryState.Stage.FINALIZE);
onGoingRecoveries.markRecoveryAsDone(recoveryId);
return null;
}), indexShard::preRecovery);
return;
}

record StartRecoveryRequestToSend(StartRecoveryRequest startRecoveryRequest, String actionName, TransportRequest requestToSend) {}
final ActionListener<StartRecoveryRequestToSend> toSendListener = failureHandler.map(r -> {
final ActionListener<StartRecoveryRequestToSend> toSendListener = cleanupOnly.map(r -> {
logger.trace(
"{} [{}]: recovery from {}",
r.startRecoveryRequest().shardId(),
Expand All @@ -273,30 +269,26 @@ record StartRecoveryRequestToSend(StartRecoveryRequest startRecoveryRequest, Str
});

if (preExistingRequest == null) {
try {
indexShard.preRecovery(toSendListener.delegateFailure((l, v) -> ActionListener.completeWith(l, () -> {
logger.trace("{} preparing shard for peer recovery", recoveryTarget.shardId());
indexShard.prepareForIndexRecovery();
if (indexShard.indexSettings().getIndexMetadata().isSearchableSnapshot()) {
// for searchable snapshots, peer recovery is treated similarly to recovery from snapshot
indexShard.getIndexEventListener().afterFilesRestoredFromRepository(indexShard);
final Store store = indexShard.store();
store.incRef();
try {
StoreRecovery.bootstrap(indexShard, store);
} finally {
store.decRef();
}
ActionListener.run(toSendListener.map(v -> {
logger.trace("{} preparing shard for peer recovery", recoveryTarget.shardId());
indexShard.prepareForIndexRecovery();
if (indexShard.indexSettings().getIndexMetadata().isSearchableSnapshot()) {
// for searchable snapshots, peer recovery is treated similarly to recovery from snapshot
indexShard.getIndexEventListener().afterFilesRestoredFromRepository(indexShard);
final Store store = indexShard.store();
store.incRef();
try {
StoreRecovery.bootstrap(indexShard, store);
} finally {
store.decRef();
}
final long startingSeqNo = indexShard.recoverLocallyUpToGlobalCheckpoint();
assert startingSeqNo == UNASSIGNED_SEQ_NO || recoveryTarget.state().getStage() == RecoveryState.Stage.TRANSLOG
: "unexpected recovery stage [" + recoveryTarget.state().getStage() + "] starting seqno [ " + startingSeqNo + "]";
final var startRequest = getStartRecoveryRequest(logger, clusterService.localNode(), recoveryTarget, startingSeqNo);
return new StartRecoveryRequestToSend(startRequest, PeerRecoverySourceService.Actions.START_RECOVERY, startRequest);
})));
} catch (Exception e) {
toSendListener.onFailure(e);
}
}
final long startingSeqNo = indexShard.recoverLocallyUpToGlobalCheckpoint();
assert startingSeqNo == UNASSIGNED_SEQ_NO || recoveryTarget.state().getStage() == RecoveryState.Stage.TRANSLOG
: "unexpected recovery stage [" + recoveryTarget.state().getStage() + "] starting seqno [ " + startingSeqNo + "]";
final var startRequest = getStartRecoveryRequest(logger, clusterService.localNode(), recoveryTarget, startingSeqNo);
return new StartRecoveryRequestToSend(startRequest, PeerRecoverySourceService.Actions.START_RECOVERY, startRequest);
}), indexShard::preRecovery);
} else {
toSendListener.onResponse(
new StartRecoveryRequestToSend(
Expand Down

0 comments on commit 134f51b

Please sign in to comment.