honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

flush pending listeners if remote cluster connection is closed

This commit is contained in:
Simon Willnauer 2017-01-12 00:21:46 +01:00
parent 00781d24ce
commit bf15decf20
2 changed files with 22 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
core/src/main/java/org/elasticsearch/action/search/RemoteClusterConnection.java
View File

@ -410,6 +410,7 @@ final class RemoteClusterConnection extends AbstractComponent implements Transpo
cancellableThreads.cancel("connect handler is closed"); cancellableThreads.cancel("connect handler is closed");
running.acquire(); // acquire the semaphore to ensure all connections are closed and all thread joined running.acquire(); // acquire the semaphore to ensure all connections are closed and all thread joined
running.release(); running.release();
maybeConnect(); // now go an notify pending listeners
} }
} catch (InterruptedException e) { } catch (InterruptedException e) {
Thread.currentThread().interrupt(); Thread.currentThread().interrupt();

22
core/src/test/java/org/elasticsearch/action/search/RemoteClusterConnectionTests.java
View File

@ -464,13 +464,23 @@ public class RemoteClusterConnectionTests extends ESTestCase {
barrier.await(); barrier.await();
CountDownLatch latch = new CountDownLatch(numConnectionAttempts); CountDownLatch latch = new CountDownLatch(numConnectionAttempts);
for (int i = 0; i < numConnectionAttempts; i++) { for (int i = 0; i < numConnectionAttempts; i++) {
AtomicBoolean executed = new AtomicBoolean(false); AtomicReference<RuntimeException> executed = new AtomicReference<>();
ActionListener<Void> listener = ActionListener.wrap(x -> { ActionListener<Void> listener = ActionListener.wrap(
assertTrue(executed.compareAndSet(false, true)); x -> {
latch.countDown();}, x -> { if (executed.compareAndSet(null, new RuntimeException())) {
assertTrue(executed.compareAndSet(false, true));
latch.countDown(); latch.countDown();
if (x instanceof RejectedExecutionException || x instanceof AlreadyClosedException) { } else {
throw new AssertionError("shit's been called twice", executed.get());
}
},
x -> {
if (executed.compareAndSet(null, new RuntimeException())) {
latch.countDown();
} else {
throw new AssertionError("shit's been called twice", executed.get());
}
if (x instanceof RejectedExecutionException || x instanceof AlreadyClosedException
|| x instanceof CancellableThreads.ExecutionCancelledException) {
// that's fine // that's fine
} else { } else {
throw new AssertionError(x); throw new AssertionError(x);