-
Notifications
You must be signed in to change notification settings - Fork 24.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add get file chunk timeouts with listener timeouts (#38758)
This commit adds a `ListenerTimeouts` class that will wrap a `ActionListener` in a listener with a timeout scheduled on the generic thread pool. If the timeout expires before the listener is completed, `onFailure` will be called with an `ElasticsearchTimeoutException`. Timeouts for the get ccr file chunk action are implemented using this functionality. Additionally, this commit attempts to fix #38027 by also blocking proxied get ccr file chunk actions. This test being un-muted is useful to verify the timeout functionality.
- Loading branch information
1 parent
3b4b80a
commit cf82864
Showing
4 changed files
with
245 additions
and
30 deletions.
There are no files selected for viewing
89 changes: 89 additions & 0 deletions
89
server/src/main/java/org/elasticsearch/action/support/ListenerTimeouts.java
This file contains 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 |
|---|---|---|
| @@ -0,0 +1,89 @@ | ||
| /* | ||
| * Licensed to Elasticsearch under one or more contributor | ||
| * license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright | ||
| * ownership. Elasticsearch licenses this file to you under | ||
| * the Apache License, Version 2.0 (the "License"); you may | ||
| * not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
|
|
||
| package org.elasticsearch.action.support; | ||
|
|
||
| import org.elasticsearch.ElasticsearchTimeoutException; | ||
| import org.elasticsearch.action.ActionListener; | ||
| import org.elasticsearch.common.unit.TimeValue; | ||
| import org.elasticsearch.threadpool.Scheduler; | ||
| import org.elasticsearch.threadpool.ThreadPool; | ||
|
|
||
| import java.util.concurrent.atomic.AtomicBoolean; | ||
|
|
||
| public class ListenerTimeouts { | ||
|
|
||
| /** | ||
| * Wraps a listener with a listener that can timeout. After the timeout period the | ||
| * {@link ActionListener#onFailure(Exception)} will be called with a | ||
| * {@link ElasticsearchTimeoutException} if the listener has not already been completed. | ||
| * | ||
| * @param threadPool used to schedule the timeout | ||
| * @param listener to that can timeout | ||
| * @param timeout period before listener failed | ||
| * @param executor to use for scheduling timeout | ||
| * @param listenerName name of the listener for timeout exception | ||
| * @return the wrapped listener that will timeout | ||
| */ | ||
| public static <Response> ActionListener<Response> wrapWithTimeout(ThreadPool threadPool, ActionListener<Response> listener, | ||
| TimeValue timeout, String executor, String listenerName) { | ||
| TimeoutableListener<Response> wrappedListener = new TimeoutableListener<>(listener, timeout, listenerName); | ||
| wrappedListener.cancellable = threadPool.schedule(wrappedListener, timeout, executor); | ||
| return wrappedListener; | ||
| } | ||
|
|
||
| private static class TimeoutableListener<Response> implements ActionListener<Response>, Runnable { | ||
|
|
||
| private final AtomicBoolean isDone = new AtomicBoolean(false); | ||
| private final ActionListener<Response> delegate; | ||
| private final TimeValue timeout; | ||
| private final String listenerName; | ||
| private volatile Scheduler.ScheduledCancellable cancellable; | ||
|
|
||
| private TimeoutableListener(ActionListener<Response> delegate, TimeValue timeout, String listenerName) { | ||
| this.delegate = delegate; | ||
| this.timeout = timeout; | ||
| this.listenerName = listenerName; | ||
| } | ||
|
|
||
| @Override | ||
| public void onResponse(Response response) { | ||
| if (isDone.compareAndSet(false, true)) { | ||
| cancellable.cancel(); | ||
| delegate.onResponse(response); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void onFailure(Exception e) { | ||
| if (isDone.compareAndSet(false, true)) { | ||
| cancellable.cancel(); | ||
| delegate.onFailure(e); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void run() { | ||
| if (isDone.compareAndSet(false, true)) { | ||
| String timeoutMessage = "[" + listenerName + "]" + " timed out after [" + timeout + "]"; | ||
| delegate.onFailure(new ElasticsearchTimeoutException(timeoutMessage)); | ||
| } | ||
| } | ||
| } | ||
| } |
120 changes: 120 additions & 0 deletions
120
server/src/test/java/org/elasticsearch/action/support/ListenerTimeoutsTests.java
This file contains 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 |
|---|---|---|
| @@ -0,0 +1,120 @@ | ||
| /* | ||
| * Licensed to Elasticsearch under one or more contributor | ||
| * license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright | ||
| * ownership. Elasticsearch licenses this file to you under | ||
| * the Apache License, Version 2.0 (the "License"); you may | ||
| * not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
|
|
||
| package org.elasticsearch.action.support; | ||
|
|
||
| import org.elasticsearch.ElasticsearchTimeoutException; | ||
| import org.elasticsearch.action.ActionListener; | ||
| import org.elasticsearch.cluster.coordination.DeterministicTaskQueue; | ||
| import org.elasticsearch.common.settings.Settings; | ||
| import org.elasticsearch.common.unit.TimeValue; | ||
| import org.elasticsearch.test.ESTestCase; | ||
| import org.elasticsearch.threadpool.ThreadPool; | ||
| import org.junit.Before; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.concurrent.atomic.AtomicBoolean; | ||
| import java.util.concurrent.atomic.AtomicReference; | ||
|
|
||
| import static org.elasticsearch.node.Node.NODE_NAME_SETTING; | ||
| import static org.hamcrest.core.IsInstanceOf.instanceOf; | ||
|
|
||
| public class ListenerTimeoutsTests extends ESTestCase { | ||
|
|
||
| private final TimeValue timeout = TimeValue.timeValueMillis(10); | ||
| private final String generic = ThreadPool.Names.GENERIC; | ||
| private DeterministicTaskQueue taskQueue; | ||
|
|
||
| @Before | ||
| public void setUp() throws Exception { | ||
| super.setUp(); | ||
| Settings settings = Settings.builder().put(NODE_NAME_SETTING.getKey(), "node").build(); | ||
| taskQueue = new DeterministicTaskQueue(settings, random()); | ||
| } | ||
|
|
||
| public void testListenerTimeout() { | ||
| AtomicBoolean success = new AtomicBoolean(false); | ||
| AtomicReference<Exception> exception = new AtomicReference<>(); | ||
| ActionListener<Void> listener = wrap(success, exception); | ||
|
|
||
| ActionListener<Void> wrapped = ListenerTimeouts.wrapWithTimeout(taskQueue.getThreadPool(), listener, timeout, generic, "test"); | ||
| assertTrue(taskQueue.hasDeferredTasks()); | ||
| taskQueue.advanceTime(); | ||
| taskQueue.runAllRunnableTasks(); | ||
|
|
||
| wrapped.onResponse(null); | ||
| wrapped.onFailure(new IOException("incorrect exception")); | ||
|
|
||
| assertFalse(success.get()); | ||
| assertThat(exception.get(), instanceOf(ElasticsearchTimeoutException.class)); | ||
| } | ||
|
|
||
| public void testFinishNormallyBeforeTimeout() { | ||
| AtomicBoolean success = new AtomicBoolean(false); | ||
| AtomicReference<Exception> exception = new AtomicReference<>(); | ||
| ActionListener<Void> listener = wrap(success, exception); | ||
|
|
||
| ActionListener<Void> wrapped = ListenerTimeouts.wrapWithTimeout(taskQueue.getThreadPool(), listener, timeout, generic, "test"); | ||
| wrapped.onResponse(null); | ||
| wrapped.onFailure(new IOException("boom")); | ||
| wrapped.onResponse(null); | ||
|
|
||
| assertTrue(taskQueue.hasDeferredTasks()); | ||
| taskQueue.advanceTime(); | ||
| taskQueue.runAllRunnableTasks(); | ||
|
|
||
| assertTrue(success.get()); | ||
| assertNull(exception.get()); | ||
| } | ||
|
|
||
| public void testFinishExceptionallyBeforeTimeout() { | ||
| AtomicBoolean success = new AtomicBoolean(false); | ||
| AtomicReference<Exception> exception = new AtomicReference<>(); | ||
| ActionListener<Void> listener = wrap(success, exception); | ||
|
|
||
| ActionListener<Void> wrapped = ListenerTimeouts.wrapWithTimeout(taskQueue.getThreadPool(), listener, timeout, generic, "test"); | ||
| wrapped.onFailure(new IOException("boom")); | ||
|
|
||
| assertTrue(taskQueue.hasDeferredTasks()); | ||
| taskQueue.advanceTime(); | ||
| taskQueue.runAllRunnableTasks(); | ||
|
|
||
| assertFalse(success.get()); | ||
| assertThat(exception.get(), instanceOf(IOException.class)); | ||
| } | ||
|
|
||
| private ActionListener<Void> wrap(AtomicBoolean success, AtomicReference<Exception> exception) { | ||
| return new ActionListener<Void>() { | ||
|
|
||
| private final AtomicBoolean completed = new AtomicBoolean(); | ||
|
|
||
| @Override | ||
| public void onResponse(Void aVoid) { | ||
| assertTrue(completed.compareAndSet(false, true)); | ||
| assertTrue(success.compareAndSet(false, true)); | ||
| } | ||
|
|
||
| @Override | ||
| public void onFailure(Exception e) { | ||
| assertTrue(completed.compareAndSet(false, true)); | ||
| assertTrue(exception.compareAndSet(null, e)); | ||
| } | ||
| }; | ||
| } | ||
| } |
This file contains 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 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