Skip to content
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

Preventing exceptions on node shutdown in integration tests #88827

Merged
merged 6 commits into from
Jul 28, 2022
Merged

Preventing exceptions on node shutdown in integration tests #88827

Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
c9ed1ee
Preventing exceptions on node shutdown
masseyke Jul 26, 2022
41b17f4
avoiding UnsupportedOperationException
masseyke Jul 26, 2022
0a6202f
checking for shutdown
masseyke Jul 27, 2022
9f8f56a
removing unneeded check
masseyke Jul 27, 2022
4b864a2
Catching EsRejectedExecutionException
masseyke Jul 27, 2022
0b7a673
restoring accidental delete
masseyke Jul 27, 2022
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
42 changes: 30 additions & 12 deletions .../src/main/java/org/elasticsearch/cluster/coordination/CoordinationDiagnosticsService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.io.stream.Writeable;
import org.elasticsearch.common.settings.Setting;
import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
import org.elasticsearch.core.Nullable;
import org.elasticsearch.core.Releasable;
import org.elasticsearch.core.Releasables;
Expand Down Expand Up @@ -485,12 +486,20 @@ void beginPollingClusterFormationInfo(
) {
masterEligibleNodes.forEach(masterEligibleNode -> {
Consumer<ClusterFormationStateOrException> responseConsumer = result -> nodeResponseConsumer.accept(masterEligibleNode, result);
cancellableConsumer.accept(
fetchClusterFormationInfo(
masterEligibleNode,
responseConsumer.andThen(rescheduleFetchConsumer(masterEligibleNode, responseConsumer, cancellableConsumer))
)
);
try {
cancellableConsumer.accept(
fetchClusterFormationInfo(
masterEligibleNode,
responseConsumer.andThen(rescheduleFetchConsumer(masterEligibleNode, responseConsumer, cancellableConsumer))
)
);
} catch (EsRejectedExecutionException e) {
if (e.isExecutorShutdown()) {
logger.trace("Not rescheduling request for cluster coordination info because this node is being shutdown", e);
} else {
throw e;
}
}
});
}

Expand All @@ -508,12 +517,20 @@ private Consumer<CoordinationDiagnosticsService.ClusterFormationStateOrException
Consumer<Scheduler.Cancellable> cancellableConsumer
) {
return response -> {
cancellableConsumer.accept(
fetchClusterFormationInfo(
masterEligibleNode,
responseConsumer.andThen(rescheduleFetchConsumer(masterEligibleNode, responseConsumer, cancellableConsumer))
)
);
try {
cancellableConsumer.accept(
fetchClusterFormationInfo(
masterEligibleNode,
responseConsumer.andThen(rescheduleFetchConsumer(masterEligibleNode, responseConsumer, cancellableConsumer))
)
);
} catch (EsRejectedExecutionException e) {
if (e.isExecutorShutdown()) {
logger.trace("Not rescheduling request for cluster coordination info because this node is being shutdown", e);
} else {
throw e;
}
}
};
}

Expand All @@ -532,6 +549,7 @@ private void cancelPollingClusterFormationInfo() {
* @param node The node to poll for cluster formation information
* @param responseConsumer The consumer of the cluster formation info for the node, or the exception encountered while contacting it
* @return A Cancellable for the task that is scheduled to fetch cluster formation information
* @throws EsRejectedExecutionException If the task cannot be scheduled, possibly because the node is shutting down.
*/
private Scheduler.Cancellable fetchClusterFormationInfo(
DiscoveryNode node,
Expand Down