Wait for new master when failing shard #15748
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -22,9 +22,11 @@ | |
| import org.elasticsearch.ExceptionsHelper; | ||
| import org.elasticsearch.cluster.ClusterService; | ||
| import org.elasticsearch.cluster.ClusterState; | ||
| import org.elasticsearch.cluster.ClusterStateObserver; | ||
| import org.elasticsearch.cluster.ClusterStateTaskConfig; | ||
| import org.elasticsearch.cluster.ClusterStateTaskExecutor; | ||
| import org.elasticsearch.cluster.ClusterStateTaskListener; | ||
| import org.elasticsearch.cluster.MasterNodeChangePredicate; | ||
| import org.elasticsearch.cluster.NotMasterException; | ||
| import org.elasticsearch.cluster.metadata.IndexMetaData; | ||
| import org.elasticsearch.cluster.node.DiscoveryNode; | ||
|
|
@@ -42,73 +44,118 @@ | |
| import org.elasticsearch.common.logging.ESLogger; | ||
| import org.elasticsearch.common.settings.Settings; | ||
| import org.elasticsearch.common.unit.TimeValue; | ||
| import org.elasticsearch.discovery.Discovery; | ||
| import org.elasticsearch.node.NodeClosedException; | ||
| import org.elasticsearch.threadpool.ThreadPool; | ||
| import org.elasticsearch.transport.EmptyTransportResponseHandler; | ||
| import org.elasticsearch.transport.NodeDisconnectedException; | ||
| import org.elasticsearch.transport.TransportChannel; | ||
| import org.elasticsearch.transport.TransportException; | ||
| import org.elasticsearch.transport.TransportRequest; | ||
| import org.elasticsearch.transport.TransportRequestHandler; | ||
| import org.elasticsearch.transport.TransportResponse; | ||
| import org.elasticsearch.transport.TransportService; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.ArrayList; | ||
| import java.util.Arrays; | ||
| import java.util.HashSet; | ||
| import java.util.List; | ||
| import java.util.Locale; | ||
| import java.util.Set; | ||
|
|
||
| import static org.elasticsearch.cluster.routing.ShardRouting.readShardRoutingEntry; | ||
|
|
||
| public class ShardStateAction extends AbstractComponent { | ||
|
|
||
| public static final String SHARD_STARTED_ACTION_NAME = "internal:cluster/shard/started"; | ||
| public static final String SHARD_FAILED_ACTION_NAME = "internal:cluster/shard/failure"; | ||
|
|
||
| private final TransportService transportService; | ||
| private final ClusterService clusterService; | ||
|
|
||
| @Inject | ||
| public ShardStateAction(Settings settings, ClusterService clusterService, TransportService transportService, | ||
| AllocationService allocationService, RoutingService routingService) { | ||
| super(settings); | ||
| this.transportService = transportService; | ||
| this.clusterService = clusterService; | ||
|
|
||
| transportService.registerRequestHandler(SHARD_STARTED_ACTION_NAME, ShardRoutingEntry::new, ThreadPool.Names.SAME, new ShardStartedTransportHandler(clusterService, new ShardStartedClusterStateTaskExecutor(allocationService, logger), logger)); | ||
| transportService.registerRequestHandler(SHARD_FAILED_ACTION_NAME, ShardRoutingEntry::new, ThreadPool.Names.SAME, new ShardFailedTransportHandler(clusterService, new ShardFailedClusterStateTaskExecutor(allocationService, routingService, logger), logger)); | ||
| } | ||
|
|
||
| public void shardFailed(final ShardRouting shardRouting, final String indexUUID, final String message, @Nullable final Throwable failure, Listener listener) { | ||
| ClusterStateObserver observer = new ClusterStateObserver(clusterService, null, logger); | ||
| ShardRoutingEntry shardRoutingEntry = new ShardRoutingEntry(shardRouting, indexUUID, message, failure); | ||
| sendShardFailed(observer, shardRoutingEntry, listener); | ||
| } | ||
|
|
||
| public void resendShardFailed(final ShardRouting shardRouting, final String indexUUID, final String message, @Nullable final Throwable failure, Listener listener) { | ||
| logger.trace("{} re-sending failed shard [{}], index UUID [{}], reason [{}]", shardRouting.shardId(), failure, shardRouting, indexUUID, message); | ||
| shardFailed(shardRouting, indexUUID, message, failure, listener); | ||
| } | ||
|
|
||
| private void sendShardFailed(ClusterStateObserver observer, ShardRoutingEntry shardRoutingEntry, Listener listener) { | ||
| DiscoveryNode masterNode = observer.observedState().nodes().masterNode(); | ||
| if (masterNode == null) { | ||
| logger.warn("{} no master known to fail shard [{}]", shardRoutingEntry.getShardRouting().shardId(), shardRoutingEntry.getShardRouting()); | ||
| waitForNewMasterAndRetry(observer, shardRoutingEntry, listener); | ||
| } else { | ||
| transportService.sendRequest(masterNode, | ||
| SHARD_FAILED_ACTION_NAME, shardRoutingEntry, new EmptyTransportResponseHandler(ThreadPool.Names.SAME) { | ||
| @Override | ||
| public void handleResponse(TransportResponse.Empty response) { | ||
| listener.onSuccess(); | ||
| } | ||
|
|
||
| @Override | ||
| public void handleException(TransportException exp) { | ||
| assert exp.getCause() != null : exp; | ||
| if (isMasterChannelException(exp.getCause())) { | ||
| waitForNewMasterAndRetry(observer, shardRoutingEntry, listener); | ||
| } else { | ||
| logger.warn("{} unexpected failure while sending request to [{}] to fail shard [{}]", exp, shardRoutingEntry.getShardRouting().shardId(), masterNode, shardRoutingEntry); | ||
| listener.onShardFailedFailure(exp); | ||
| } | ||
| } | ||
| }); | ||
| } | ||
| } | ||
|
|
||
| private static Set<Class<?>> MASTER_CHANNEL_EXCEPTIONS = | ||
| new HashSet<>(Arrays.asList( | ||
| NotMasterException.class, | ||
| NodeDisconnectedException.class, | ||
| Discovery.FailedToCommitClusterStateException.class | ||
| )); | ||
| private static boolean isMasterChannelException(Throwable cause) { | ||
| return MASTER_CHANNEL_EXCEPTIONS.contains(cause.getClass()); | ||
|
Contributor
There was a problem hiding this comment. don't we need to unwrap the cause (it's a TransportException now) ? If so, I think we need to strengthen our IT tests here..
Member
Author
There was a problem hiding this comment. @bleskes The cause is unwrapped in |
||
| } | ||
|
|
||
| // visible for testing | ||
| protected void waitForNewMasterAndRetry(ClusterStateObserver observer, ShardRoutingEntry shardRoutingEntry, Listener listener) { | ||
| observer.waitForNextChange(new ClusterStateObserver.Listener() { | ||
| @Override | ||
| public void onNewClusterState(ClusterState state) { | ||
| if (logger.isTraceEnabled()) { | ||
| logger.trace("new cluster state [{}] after waiting for master election to fail shard [{}]", shardRoutingEntry.getShardRouting().shardId(), state.prettyPrint(), shardRoutingEntry); | ||
| } | ||
| sendShardFailed(observer, shardRoutingEntry, listener); | ||
|
Contributor
There was a problem hiding this comment. can we add a trace log here? |
||
| } | ||
|
|
||
| @Override | ||
| public void onClusterServiceClose() { | ||
| logger.warn("{} node closed while handling failed shard [{}]", shardRoutingEntry.failure, shardRoutingEntry.getShardRouting().getId(), shardRoutingEntry.getShardRouting()); | ||
| listener.onShardFailedFailure(new NodeClosedException(clusterService.localNode())); | ||
| } | ||
|
|
||
| @Override | ||
| public void onTimeout(TimeValue timeout) { | ||
| // we wait indefinitely for a new master | ||
| assert false; | ||
| } | ||
| }, MasterNodeChangePredicate.INSTANCE); | ||
| } | ||
|
|
||
| private static class ShardFailedTransportHandler implements TransportRequestHandler<ShardRoutingEntry> { | ||
|
|
@@ -334,10 +381,22 @@ public interface Listener { | |
| default void onSuccess() { | ||
| } | ||
|
|
||
| /** | ||
| * Notification for non-channel exceptions that are not handled | ||
| * by {@link ShardStateAction}. | ||
| * | ||
| * The exceptions that are handled by {@link ShardStateAction} | ||
| * are: | ||
| * - {@link NotMasterException} | ||
| * - {@link NodeDisconnectedException} | ||
| * - {@link Discovery.FailedToCommitClusterStateException} | ||
| * | ||
| * Any other exception is communicated to the requester via | ||
| * this notification. | ||
| * | ||
| * @param e the unexpected cause of the failure on the master | ||
| */ | ||
| default void onShardFailedFailure(final Exception e) { | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thank you.