Skip to content

Fix Snapshot Completion Listener Lost on Master Failover (#54286) #54330

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Mar 27, 2020
Merged

Fix Snapshot Completion Listener Lost on Master Failover (#54286) #54330

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 27 additions & 12 deletions server/src/main/java/org/elasticsearch/snapshots/SnapshotsService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1104,7 +1104,8 @@ public void onFailure(final Exception e) {
// will try ending this snapshot again
logger.debug(() -> new ParameterizedMessage(
"[{}] failed to update cluster state during snapshot finalization", snapshot), e);
endingSnapshots.remove(snapshot);
failSnapshotCompletionListeners(snapshot,
new SnapshotException(snapshot, "Failed to update cluster state during snapshot finalization", e));
} else {
logger.warn(() -> new ParameterizedMessage("[{}] failed to finalize snapshot", snapshot), e);
removeSnapshotFromClusterState(snapshot, null, e);
Expand Down Expand Up @@ -1158,42 +1159,56 @@ public ClusterState execute(ClusterState currentState) {
@Override
public void onFailure(String source, Exception e) {
logger.warn(() -> new ParameterizedMessage("[{}] failed to remove snapshot metadata", snapshot), e);
endingSnapshots.remove(snapshot);
failSnapshotCompletionListeners(
snapshot, new SnapshotException(snapshot, "Failed to remove snapshot from cluster state", e));
if (listener != null) {
listener.onFailure(e);
}
}

@Override
public void onNoLongerMaster(String source) {
endingSnapshots.remove(snapshot);
failSnapshotCompletionListeners(
snapshot, ExceptionsHelper.useOrSuppress(failure, new SnapshotException(snapshot, "no longer master")));
if (listener != null) {
listener.onNoLongerMaster();
}
}

@Override
public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) {
final List<ActionListener<SnapshotInfo>> completionListeners = snapshotCompletionListeners.remove(snapshot);
if (completionListeners != null) {
try {
if (snapshotInfo == null) {
ActionListener.onFailure(completionListeners, failure);
} else {
if (snapshotInfo == null) {
failSnapshotCompletionListeners(snapshot, failure);
} else {
final List<ActionListener<SnapshotInfo>> completionListeners = snapshotCompletionListeners.remove(snapshot);
if (completionListeners != null) {
try {
ActionListener.onResponse(completionListeners, snapshotInfo);
} catch (Exception e) {
logger.warn("Failed to notify listeners", e);
}
} catch (Exception e) {
logger.warn("Failed to notify listeners", e);
}
endingSnapshots.remove(snapshot);
}
endingSnapshots.remove(snapshot);
if (listener != null) {
listener.onResponse(snapshotInfo);
}
}
});
}

private void failSnapshotCompletionListeners(Snapshot snapshot, Exception e) {
final List<ActionListener<SnapshotInfo>> completionListeners = snapshotCompletionListeners.remove(snapshot);
if (completionListeners != null) {
try {
ActionListener.onFailure(completionListeners, e);
} catch (Exception ex) {
logger.warn("Failed to notify listeners", ex);
}
}
endingSnapshots.remove(snapshot);
}

/**
* Deletes a snapshot from the repository, looking up the {@link Snapshot} reference before deleting.
* If the snapshot is still running cancels the snapshot first and then deletes it from the repository.
Expand Down
96 changes: 96 additions & 0 deletions server/src/test/java/org/elasticsearch/discovery/SnapshotDisruptionIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
*/
package org.elasticsearch.discovery;

import org.elasticsearch.ExceptionsHelper;
import org.elasticsearch.action.ActionFuture;
import org.elasticsearch.action.admin.cluster.snapshots.create.CreateSnapshotResponse;
import org.elasticsearch.action.admin.cluster.snapshots.get.GetSnapshotsResponse;
Expand All @@ -27,11 +28,14 @@
import org.elasticsearch.cluster.ClusterStateListener;
import org.elasticsearch.cluster.SnapshotDeletionsInProgress;
import org.elasticsearch.cluster.SnapshotsInProgress;
import org.elasticsearch.cluster.metadata.RepositoriesMetaData;
import org.elasticsearch.cluster.metadata.RepositoryMetaData;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.plugins.Plugin;
import org.elasticsearch.snapshots.ConcurrentSnapshotExecutionException;
import org.elasticsearch.snapshots.SnapshotException;
import org.elasticsearch.snapshots.SnapshotInfo;
import org.elasticsearch.snapshots.SnapshotMissingException;
import org.elasticsearch.snapshots.SnapshotState;
Expand All @@ -49,6 +53,9 @@
import java.util.concurrent.TimeUnit;

import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.hamcrest.Matchers.either;
import static org.hamcrest.Matchers.endsWith;
import static org.hamcrest.Matchers.is;

/**
* Tests snapshot operations during disruptions.
Expand Down Expand Up @@ -156,6 +163,95 @@ public void clusterChanged(ClusterChangedEvent event) {
assertAllSnapshotsCompleted();
}

public void testDisruptionAfterFinalization() throws Exception {
final String idxName = "test";
final List<String> allMasterEligibleNodes = internalCluster().startMasterOnlyNodes(3);
final String dataNode = internalCluster().startDataOnlyNode();
ensureStableCluster(4);

createRandomIndex(idxName);

logger.info("--> creating repository");
assertAcked(client().admin().cluster().preparePutRepository("test-repo")
.setType("fs").setSettings(Settings.builder()
.put("location", randomRepoPath())
.put("compress", randomBoolean())
.put("chunk_size", randomIntBetween(100, 1000), ByteSizeUnit.BYTES)));

final String masterNode1 = internalCluster().getMasterName();
Set<String> otherNodes = new HashSet<>(allMasterEligibleNodes);
otherNodes.remove(masterNode1);
otherNodes.add(dataNode);

NetworkDisruption networkDisruption =
new NetworkDisruption(new NetworkDisruption.TwoPartitions(Collections.singleton(masterNode1), otherNodes),
new NetworkDisruption.NetworkUnresponsive());
internalCluster().setDisruptionScheme(networkDisruption);

ClusterService clusterService = internalCluster().clusterService(masterNode1);
CountDownLatch disruptionStarted = new CountDownLatch(1);
clusterService.addListener(new ClusterStateListener() {
@Override
public void clusterChanged(ClusterChangedEvent event) {
SnapshotsInProgress snapshots = event.state().custom(SnapshotsInProgress.TYPE);
if (snapshots != null && snapshots.entries().size() > 0) {
final SnapshotsInProgress.Entry snapshotEntry = snapshots.entries().get(0);
if (snapshotEntry.state() == SnapshotsInProgress.State.SUCCESS) {
final RepositoriesMetaData repoMeta =
event.state().metaData().custom(RepositoriesMetaData.TYPE);
final RepositoryMetaData metaData = repoMeta.repository("test-repo");
if (metaData.generation() == metaData.pendingGeneration()
&& metaData.generation() > snapshotEntry.repositoryStateId()) {
logger.info("--> starting disruption");
networkDisruption.startDisrupting();
clusterService.removeListener(this);
disruptionStarted.countDown();
}
}
}
}
});

final String snapshot = "test-snap";

logger.info("--> starting snapshot");
ActionFuture<CreateSnapshotResponse> future = client(masterNode1).admin().cluster()
.prepareCreateSnapshot("test-repo", snapshot).setWaitForCompletion(true)
.setIndices(idxName).execute();

logger.info("--> waiting for disruption to start");
assertTrue(disruptionStarted.await(1, TimeUnit.MINUTES));

assertAllSnapshotsCompleted();

logger.info("--> verify that snapshot was successful or no longer exist");
assertBusy(() -> {
try {
assertSnapshotExists("test-repo", snapshot);
} catch (SnapshotMissingException exception) {
logger.info("--> done verifying, snapshot doesn't exist");
}
}, 1, TimeUnit.MINUTES);

logger.info("--> stopping disrupting");
networkDisruption.stopDisrupting();
ensureStableCluster(4, masterNode1);
logger.info("--> done");

try {
future.get();
fail("Should have failed because the node disconnected from the cluster during snapshot finalization");
} catch (Exception ex) {
final SnapshotException sne = (SnapshotException) ExceptionsHelper.unwrap(ex, SnapshotException.class);
assertNotNull(sne);
assertThat(
sne.getMessage(), either(endsWith(" Failed to remove snapshot from cluster state")).or(endsWith(" no longer master")));
assertThat(sne.getSnapshotName(), is(snapshot));
}

assertAllSnapshotsCompleted();
}

private void assertAllSnapshotsCompleted() throws Exception {
logger.info("--> wait until the snapshot is done");
assertBusy(() -> {
Expand Down