From 0260c6f55c4623f1195341bdc4e61a7ec7bc1754 Mon Sep 17 00:00:00 2001 From: Jay Modi Date: Mon, 18 Nov 2019 14:47:32 -0700 Subject: [PATCH] ThreadPool and ThreadContext are not closeable (#43249) This commit changes the ThreadContext to just use a regular ThreadLocal over the lucene CloseableThreadLocal. The CloseableThreadLocal solves issues with ThreadLocals that are no longer needed during runtime but in the case of the ThreadContext, we need it for the runtime of the node and it is typically not closed until the node closes, so we miss out on the benefits that this class provides. Additionally by removing the close logic, we simplify code in other places that deal with exceptions and tracking to see if it happens when the node is closing. Closes #42577 --- .../common/logging/JsonLoggerTests.java | 123 +++-- .../common/logging/DeprecationLogger.java | 1 - .../util/concurrent/EsThreadPoolExecutor.java | 13 +- .../common/util/concurrent/ThreadContext.java | 166 ++----- .../elasticsearch/threadpool/ThreadPool.java | 22 +- .../transport/TransportLogger.java | 6 +- .../ContextPreservingActionListenerTests.java | 221 +++++---- .../logging/DeprecationLoggerTests.java | 247 +++++----- .../concurrent/AsyncIOProcessorTests.java | 6 - ...ueueResizingEsThreadPoolExecutorTests.java | 7 - .../util/concurrent/ThreadContextTests.java | 423 ++++++++---------- .../org/elasticsearch/node/NodeTests.java | 1 - .../threadpool/ThreadPoolTests.java | 3 +- .../transport/TransportLoggerTests.java | 5 +- .../org/elasticsearch/test/ESTestCase.java | 25 +- .../test/rest/ESRestTestCase.java | 13 +- .../slm/SnapshotLifecycleServiceTests.java | 44 +- .../slm/SnapshotRetentionServiceTests.java | 24 +- .../xpack/slm/SnapshotRetentionTaskTests.java | 38 +- .../xpack/security/SecurityTests.java | 13 +- ...curityServerTransportInterceptorTests.java | 79 ++-- 21 files changed, 641 insertions(+), 839 deletions(-) diff --git a/qa/logging-config/src/test/java/org/elasticsearch/common/logging/JsonLoggerTests.java b/qa/logging-config/src/test/java/org/elasticsearch/common/logging/JsonLoggerTests.java index 7bb22854769c..d7f0cadbdbe9 100644 --- a/qa/logging-config/src/test/java/org/elasticsearch/common/logging/JsonLoggerTests.java +++ b/qa/logging-config/src/test/java/org/elasticsearch/common/logging/JsonLoggerTests.java @@ -387,80 +387,77 @@ public class JsonLoggerTests extends ESTestCase { public void testDuplicateLogMessages() throws IOException { final DeprecationLogger deprecationLogger = new DeprecationLogger(LogManager.getLogger("test")); - // For the same key and X-Opaque-ID deprecation should be once - try (ThreadContext threadContext = new ThreadContext(Settings.EMPTY)) { - try{ - threadContext.putHeader(Task.X_OPAQUE_ID, "ID1"); - DeprecationLogger.setThreadContext(threadContext); - deprecationLogger.deprecatedAndMaybeLog("key", "message1"); - deprecationLogger.deprecatedAndMaybeLog("key", "message2"); - assertWarnings("message1", "message2"); + ThreadContext threadContext = new ThreadContext(Settings.EMPTY); + try (ThreadContext.StoredContext ignore = threadContext.stashContext()) { + threadContext.putHeader(Task.X_OPAQUE_ID, "ID1"); + DeprecationLogger.setThreadContext(threadContext); + deprecationLogger.deprecatedAndMaybeLog("key", "message1"); + deprecationLogger.deprecatedAndMaybeLog("key", "message2"); + assertWarnings("message1", "message2"); - final Path path = PathUtils.get(System.getProperty("es.logs.base_path"), - System.getProperty("es.logs.cluster_name") + "_deprecated.json"); - try (Stream> stream = JsonLogsStream.mapStreamFrom(path)) { - List> jsonLogs = stream - .collect(Collectors.toList()); + final Path path = PathUtils.get(System.getProperty("es.logs.base_path"), + System.getProperty("es.logs.cluster_name") + "_deprecated.json"); + try (Stream> stream = JsonLogsStream.mapStreamFrom(path)) { + List> jsonLogs = stream + .collect(Collectors.toList()); - assertThat(jsonLogs, contains( - allOf( - hasEntry("type", "deprecation"), - hasEntry("level", "WARN"), - hasEntry("component", "d.test"), - hasEntry("cluster.name", "elasticsearch"), - hasEntry("node.name", "sample-name"), - hasEntry("message", "message1"), - hasEntry("x-opaque-id", "ID1")) - ) - ); - } - }finally{ - DeprecationLogger.removeThreadContext(threadContext); + assertThat(jsonLogs, contains( + allOf( + hasEntry("type", "deprecation"), + hasEntry("level", "WARN"), + hasEntry("component", "d.test"), + hasEntry("cluster.name", "elasticsearch"), + hasEntry("node.name", "sample-name"), + hasEntry("message", "message1"), + hasEntry("x-opaque-id", "ID1")) + ) + ); } + } finally { + DeprecationLogger.removeThreadContext(threadContext); } + // For the same key and different X-Opaque-ID should be multiple times per key/x-opaque-id //continuing with message1-ID1 in logs already, adding a new deprecation log line with message2-ID2 - try (ThreadContext threadContext = new ThreadContext(Settings.EMPTY)) { - try{ - threadContext.putHeader(Task.X_OPAQUE_ID, "ID2"); - DeprecationLogger.setThreadContext(threadContext); - deprecationLogger.deprecatedAndMaybeLog("key", "message1"); - deprecationLogger.deprecatedAndMaybeLog("key", "message2"); - assertWarnings("message1", "message2"); + try (ThreadContext.StoredContext ignore = threadContext.stashContext()) { + threadContext.putHeader(Task.X_OPAQUE_ID, "ID2"); + DeprecationLogger.setThreadContext(threadContext); + deprecationLogger.deprecatedAndMaybeLog("key", "message1"); + deprecationLogger.deprecatedAndMaybeLog("key", "message2"); + assertWarnings("message1", "message2"); - final Path path = PathUtils.get(System.getProperty("es.logs.base_path"), - System.getProperty("es.logs.cluster_name") + "_deprecated.json"); - try (Stream> stream = JsonLogsStream.mapStreamFrom(path)) { - List> jsonLogs = stream - .collect(Collectors.toList()); + final Path path = PathUtils.get(System.getProperty("es.logs.base_path"), + System.getProperty("es.logs.cluster_name") + "_deprecated.json"); + try (Stream> stream = JsonLogsStream.mapStreamFrom(path)) { + List> jsonLogs = stream + .collect(Collectors.toList()); - assertThat(jsonLogs, contains( - allOf( - hasEntry("type", "deprecation"), - hasEntry("level", "WARN"), - hasEntry("component", "d.test"), - hasEntry("cluster.name", "elasticsearch"), - hasEntry("node.name", "sample-name"), - hasEntry("message", "message1"), - hasEntry("x-opaque-id", "ID1") - ), - allOf( - hasEntry("type", "deprecation"), - hasEntry("level", "WARN"), - hasEntry("component", "d.test"), - hasEntry("cluster.name", "elasticsearch"), - hasEntry("node.name", "sample-name"), - hasEntry("message", "message1"), - hasEntry("x-opaque-id", "ID2") - ) - ) - ); - } - }finally{ - DeprecationLogger.removeThreadContext(threadContext); + assertThat(jsonLogs, contains( + allOf( + hasEntry("type", "deprecation"), + hasEntry("level", "WARN"), + hasEntry("component", "d.test"), + hasEntry("cluster.name", "elasticsearch"), + hasEntry("node.name", "sample-name"), + hasEntry("message", "message1"), + hasEntry("x-opaque-id", "ID1") + ), + allOf( + hasEntry("type", "deprecation"), + hasEntry("level", "WARN"), + hasEntry("component", "d.test"), + hasEntry("cluster.name", "elasticsearch"), + hasEntry("node.name", "sample-name"), + hasEntry("message", "message1"), + hasEntry("x-opaque-id", "ID2") + ) + ) + ); } + } finally { + DeprecationLogger.removeThreadContext(threadContext); } } diff --git a/server/src/main/java/org/elasticsearch/common/logging/DeprecationLogger.java b/server/src/main/java/org/elasticsearch/common/logging/DeprecationLogger.java index 79ef7ec0aad8..be157e16de20 100644 --- a/server/src/main/java/org/elasticsearch/common/logging/DeprecationLogger.java +++ b/server/src/main/java/org/elasticsearch/common/logging/DeprecationLogger.java @@ -262,7 +262,6 @@ public class DeprecationLogger { public String getXOpaqueId(Set threadContexts) { return threadContexts.stream() - .filter(t -> t.isClosed() == false) .filter(t -> t.getHeader(Task.X_OPAQUE_ID) != null) .findFirst() .map(t -> t.getHeader(Task.X_OPAQUE_ID)) diff --git a/server/src/main/java/org/elasticsearch/common/util/concurrent/EsThreadPoolExecutor.java b/server/src/main/java/org/elasticsearch/common/util/concurrent/EsThreadPoolExecutor.java index dafbc8f83511..4a95c34975ae 100644 --- a/server/src/main/java/org/elasticsearch/common/util/concurrent/EsThreadPoolExecutor.java +++ b/server/src/main/java/org/elasticsearch/common/util/concurrent/EsThreadPoolExecutor.java @@ -106,17 +106,8 @@ public class EsThreadPoolExecutor extends ThreadPoolExecutor { } private boolean assertDefaultContext(Runnable r) { - try { - assert contextHolder.isDefaultContext() : "the thread context is not the default context and the thread [" + - Thread.currentThread().getName() + "] is being returned to the pool after executing [" + r + "]"; - } catch (IllegalStateException ex) { - // sometimes we execute on a closed context and isDefaultContext doen't bypass the ensureOpen checks - // this must not trigger an exception here since we only assert if the default is restored and - // we don't really care if we are closed - if (contextHolder.isClosed() == false) { - throw ex; - } - } + assert contextHolder.isDefaultContext() : "the thread context is not the default context and the thread [" + + Thread.currentThread().getName() + "] is being returned to the pool after executing [" + r + "]"; return true; } diff --git a/server/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java b/server/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java index d94bf7bf410e..e4cb14857932 100644 --- a/server/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java +++ b/server/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java @@ -20,9 +20,9 @@ package org.elasticsearch.common.util.concurrent; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; -import org.apache.lucene.util.CloseableThreadLocal; import org.elasticsearch.action.support.ContextPreservingActionListener; import org.elasticsearch.client.OriginSettingClient; +import org.elasticsearch.common.collect.Tuple; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.common.io.stream.Writeable; @@ -32,7 +32,6 @@ import org.elasticsearch.common.settings.Settings; import org.elasticsearch.http.HttpTransportSettings; import org.elasticsearch.tasks.Task; -import java.io.Closeable; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.Collections; @@ -42,7 +41,6 @@ import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Set; -import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.BiConsumer; import java.util.function.BinaryOperator; import java.util.function.Function; @@ -82,7 +80,7 @@ import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_MAX_WARN * * */ -public final class ThreadContext implements Closeable, Writeable { +public final class ThreadContext implements Writeable { public static final String PREFIX = "request.headers"; public static final Setting DEFAULT_HEADERS_SETTING = Setting.groupSetting(PREFIX + ".", Property.NodeScope); @@ -95,7 +93,7 @@ public final class ThreadContext implements Closeable, Writeable { private static final Logger logger = LogManager.getLogger(ThreadContext.class); private static final ThreadContextStruct DEFAULT_CONTEXT = new ThreadContextStruct(); private final Map defaultHeader; - private final ContextThreadLocal threadLocal; + private final ThreadLocal threadLocal; private final int maxWarningHeaderCount; private final long maxWarningHeaderSize; @@ -104,26 +102,12 @@ public final class ThreadContext implements Closeable, Writeable { * @param settings the settings to read the default request headers from */ public ThreadContext(Settings settings) { - Settings headers = DEFAULT_HEADERS_SETTING.get(settings); - if (headers == null) { - this.defaultHeader = Collections.emptyMap(); - } else { - Map defaultHeader = new HashMap<>(); - for (String key : headers.names()) { - defaultHeader.put(key, headers.get(key)); - } - this.defaultHeader = Collections.unmodifiableMap(defaultHeader); - } - threadLocal = new ContextThreadLocal(); + this.defaultHeader = buildDefaultHeaders(settings); + this.threadLocal = ThreadLocal.withInitial(() -> DEFAULT_CONTEXT); this.maxWarningHeaderCount = SETTING_HTTP_MAX_WARNING_HEADER_COUNT.get(settings); this.maxWarningHeaderSize = SETTING_HTTP_MAX_WARNING_HEADER_SIZE.get(settings).getBytes(); } - @Override - public void close() { - threadLocal.close(); - } - /** * Removes the current context and resets a default context. The removed context can be * restored by closing the returned {@link StoredContext}. @@ -140,19 +124,13 @@ public final class ThreadContext implements Closeable, Writeable { DEFAULT_CONTEXT.putHeaders(Map.of(Task.X_OPAQUE_ID, context.requestHeaders.get(Task.X_OPAQUE_ID))); threadLocal.set(threadContextStruct); } else { - threadLocal.set(null); + threadLocal.set(DEFAULT_CONTEXT); } return () -> { // If the node and thus the threadLocal get closed while this task // is still executing, we don't want this runnable to fail with an // uncaught exception - try { - threadLocal.set(context); - } catch (IllegalStateException e) { - if (isClosed() == false) { - throw e; - } - } + threadLocal.set(context); }; } @@ -255,7 +233,20 @@ public final class ThreadContext implements Closeable, Writeable { * Reads the headers from the stream into the current context */ public void readHeaders(StreamInput in) throws IOException { - final Map requestHeaders = in.readMap(StreamInput::readString, StreamInput::readString); + final Tuple, Map>> streamTuple = readHeadersFromStream(in); + final Map requestHeaders = streamTuple.v1(); + final Map> responseHeaders = streamTuple.v2(); + final ThreadContextStruct struct; + if (requestHeaders.isEmpty() && responseHeaders.isEmpty()) { + struct = ThreadContextStruct.EMPTY; + } else { + struct = new ThreadContextStruct(requestHeaders, responseHeaders, Collections.emptyMap(), false); + } + threadLocal.set(struct); + } + + public static Tuple, Map>> readHeadersFromStream(StreamInput in) throws IOException { + final Map requestHeaders = in.readMap(StreamInput::readString, StreamInput::readString); final Map> responseHeaders = in.readMap(StreamInput::readString, input -> { final int size = input.readVInt(); if (size == 0) { @@ -273,13 +264,7 @@ public final class ThreadContext implements Closeable, Writeable { return values; } }); - final ThreadContextStruct struct; - if (requestHeaders.isEmpty() && responseHeaders.isEmpty()) { - struct = ThreadContextStruct.EMPTY; - } else { - struct = new ThreadContextStruct(requestHeaders, responseHeaders, Collections.emptyMap(), false); - } - threadLocal.set(struct); + return new Tuple<>(requestHeaders, responseHeaders); } /** @@ -373,17 +358,7 @@ public final class ThreadContext implements Closeable, Writeable { * @param uniqueValue the function that produces de-duplication values */ public void addResponseHeader(final String key, final String value, final Function uniqueValue) { - /* - * Updating the thread local is expensive due to a shared reference that we synchronize on, so we should only do it if the thread - * context struct changed. It will not change if we de-duplicate this value to an existing one, or if we don't add a new one because - * we have reached capacity. - */ - final ThreadContextStruct current = threadLocal.get(); - final ThreadContextStruct maybeNext = - current.putResponse(key, value, uniqueValue, maxWarningHeaderCount, maxWarningHeaderSize); - if (current != maybeNext) { - threadLocal.set(maybeNext); - } + threadLocal.set(threadLocal.get().putResponse(key, value, uniqueValue, maxWarningHeaderCount, maxWarningHeaderSize)); } /** @@ -435,13 +410,6 @@ public final class ThreadContext implements Closeable, Writeable { return threadLocal.get().isSystemContext; } - /** - * Returns true if the context is closed, otherwise true - */ - public boolean isClosed() { - return threadLocal.closed.get(); - } - @FunctionalInterface public interface StoredContext extends AutoCloseable { @Override @@ -452,6 +420,19 @@ public final class ThreadContext implements Closeable, Writeable { } } + public static Map buildDefaultHeaders(Settings settings) { + Settings headers = DEFAULT_HEADERS_SETTING.get(settings); + if (headers == null) { + return Collections.emptyMap(); + } else { + Map defaultHeader = new HashMap<>(); + for (String key : headers.names()) { + defaultHeader.put(key, headers.get(key)); + } + return Collections.unmodifiableMap(defaultHeader); + } + } + private static final class ThreadContextStruct { private static final ThreadContextStruct EMPTY = @@ -629,55 +610,6 @@ public final class ThreadContext implements Closeable, Writeable { } } - private static class ContextThreadLocal extends CloseableThreadLocal { - private final AtomicBoolean closed = new AtomicBoolean(false); - - @Override - public void set(ThreadContextStruct object) { - try { - if (object == DEFAULT_CONTEXT) { - super.set(null); - } else { - super.set(object); - } - } catch (NullPointerException ex) { - /* This is odd but CloseableThreadLocal throws a NPE if it was closed but still accessed. - to get a real exception we call ensureOpen() to tell the user we are already closed.*/ - ensureOpen(); - throw ex; - } - } - - @Override - public ThreadContextStruct get() { - try { - ThreadContextStruct threadContextStruct = super.get(); - if (threadContextStruct != null) { - return threadContextStruct; - } - return DEFAULT_CONTEXT; - } catch (NullPointerException ex) { - /* This is odd but CloseableThreadLocal throws a NPE if it was closed but still accessed. - to get a real exception we call ensureOpen() to tell the user we are already closed.*/ - ensureOpen(); - throw ex; - } - } - - private void ensureOpen() { - if (closed.get()) { - throw new IllegalStateException("threadcontext is already closed"); - } - } - - @Override - public void close() { - if (closed.compareAndSet(false, true)) { - super.close(); - } - } - } - /** * Wraps a Runnable to preserve the thread context. */ @@ -692,19 +624,9 @@ public final class ThreadContext implements Closeable, Writeable { @Override public void run() { - boolean whileRunning = false; try (ThreadContext.StoredContext ignore = stashContext()){ ctx.restore(); - whileRunning = true; in.run(); - whileRunning = false; - } catch (IllegalStateException ex) { - if (whileRunning || threadLocal.closed.get() == false) { - throw ex; - } - // if we hit an ISE here we have been shutting down - // this comes from the threadcontext and barfs if - // our threadpool has been shutting down } } @@ -761,21 +683,9 @@ public final class ThreadContext implements Closeable, Writeable { @Override protected void doRun() throws Exception { - boolean whileRunning = false; threadsOriginalContext = stashContext(); - try { - creatorsContext.restore(); - whileRunning = true; - in.doRun(); - whileRunning = false; - } catch (IllegalStateException ex) { - if (whileRunning || threadLocal.closed.get() == false) { - throw ex; - } - // if we hit an ISE here we have been shutting down - // this comes from the threadcontext and barfs if - // our threadpool has been shutting down - } + creatorsContext.restore(); + in.doRun(); } @Override diff --git a/server/src/main/java/org/elasticsearch/threadpool/ThreadPool.java b/server/src/main/java/org/elasticsearch/threadpool/ThreadPool.java index 85922de201b9..81951fd1ec8e 100644 --- a/server/src/main/java/org/elasticsearch/threadpool/ThreadPool.java +++ b/server/src/main/java/org/elasticsearch/threadpool/ThreadPool.java @@ -39,7 +39,6 @@ import org.elasticsearch.common.xcontent.ToXContentFragment; import org.elasticsearch.common.xcontent.XContentBuilder; import org.elasticsearch.node.Node; -import java.io.Closeable; import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; @@ -62,7 +61,7 @@ import java.util.stream.Collectors; import static java.util.Collections.unmodifiableMap; import static java.util.Map.entry; -public class ThreadPool implements Scheduler, Closeable { +public class ThreadPool implements Scheduler { private static final Logger logger = LogManager.getLogger(ThreadPool.class); @@ -707,15 +706,13 @@ public class ThreadPool implements Scheduler, Closeable { public static boolean terminate(ThreadPool pool, long timeout, TimeUnit timeUnit) { if (pool != null) { // Leverage try-with-resources to close the threadpool - try (ThreadPool c = pool) { - pool.shutdown(); - if (awaitTermination(pool, timeout, timeUnit)) { - return true; - } - // last resort - pool.shutdownNow(); - return awaitTermination(pool, timeout, timeUnit); + pool.shutdown(); + if (awaitTermination(pool, timeout, timeUnit)) { + return true; } + // last resort + pool.shutdownNow(); + return awaitTermination(pool, timeout, timeUnit); } return false; } @@ -734,11 +731,6 @@ public class ThreadPool implements Scheduler, Closeable { return false; } - @Override - public void close() { - threadContext.close(); - } - public ThreadContext getThreadContext() { return threadContext; } diff --git a/server/src/main/java/org/elasticsearch/transport/TransportLogger.java b/server/src/main/java/org/elasticsearch/transport/TransportLogger.java index 5e1f92fc2cd1..09f22da59563 100644 --- a/server/src/main/java/org/elasticsearch/transport/TransportLogger.java +++ b/server/src/main/java/org/elasticsearch/transport/TransportLogger.java @@ -25,7 +25,6 @@ import org.elasticsearch.common.bytes.BytesReference; import org.elasticsearch.common.compress.Compressor; import org.elasticsearch.common.compress.NotCompressedException; import org.elasticsearch.common.io.stream.StreamInput; -import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.util.concurrent.ThreadContext; import org.elasticsearch.core.internal.io.IOUtils; @@ -94,9 +93,8 @@ public final class TransportLogger { streamInput = compressor.streamInput(streamInput); } - try (ThreadContext context = new ThreadContext(Settings.EMPTY)) { - context.readHeaders(streamInput); - } + // read and discard headers + ThreadContext.readHeadersFromStream(streamInput); if (streamInput.getVersion().before(Version.V_8_0_0)) { // discard the features streamInput.readStringArray(); diff --git a/server/src/test/java/org/elasticsearch/action/support/ContextPreservingActionListenerTests.java b/server/src/test/java/org/elasticsearch/action/support/ContextPreservingActionListenerTests.java index dd627041f226..f0277c65c16d 100644 --- a/server/src/test/java/org/elasticsearch/action/support/ContextPreservingActionListenerTests.java +++ b/server/src/test/java/org/elasticsearch/action/support/ContextPreservingActionListenerTests.java @@ -28,129 +28,126 @@ import java.io.IOException; public class ContextPreservingActionListenerTests extends ESTestCase { public void testOriginalContextIsPreservedAfterOnResponse() throws IOException { - try (ThreadContext threadContext = new ThreadContext(Settings.EMPTY)) { - final boolean nonEmptyContext = randomBoolean(); - if (nonEmptyContext) { - threadContext.putHeader("not empty", "value"); - } - final ContextPreservingActionListener actionListener; - try (ThreadContext.StoredContext ignore = threadContext.stashContext()) { - threadContext.putHeader("foo", "bar"); - final ActionListener delegate = new ActionListener() { - @Override - public void onResponse(Void aVoid) { - assertEquals("bar", threadContext.getHeader("foo")); - assertNull(threadContext.getHeader("not empty")); - } - - @Override - public void onFailure(Exception e) { - throw new RuntimeException("onFailure shouldn't be called", e); - } - }; - if (randomBoolean()) { - actionListener = new ContextPreservingActionListener<>(threadContext.newRestorableContext(true), delegate); - } else { - actionListener = ContextPreservingActionListener.wrapPreservingContext(delegate, threadContext); - } - } - - assertNull(threadContext.getHeader("foo")); - assertEquals(nonEmptyContext ? "value" : null, threadContext.getHeader("not empty")); - - actionListener.onResponse(null); - - assertNull(threadContext.getHeader("foo")); - assertEquals(nonEmptyContext ? "value" : null, threadContext.getHeader("not empty")); + ThreadContext threadContext = new ThreadContext(Settings.EMPTY); + final boolean nonEmptyContext = randomBoolean(); + if (nonEmptyContext) { + threadContext.putHeader("not empty", "value"); } + final ContextPreservingActionListener actionListener; + try (ThreadContext.StoredContext ignore = threadContext.stashContext()) { + threadContext.putHeader("foo", "bar"); + final ActionListener delegate = new ActionListener() { + @Override + public void onResponse(Void aVoid) { + assertEquals("bar", threadContext.getHeader("foo")); + assertNull(threadContext.getHeader("not empty")); + } + + @Override + public void onFailure(Exception e) { + throw new RuntimeException("onFailure shouldn't be called", e); + } + }; + if (randomBoolean()) { + actionListener = new ContextPreservingActionListener<>(threadContext.newRestorableContext(true), delegate); + } else { + actionListener = ContextPreservingActionListener.wrapPreservingContext(delegate, threadContext); + } + } + + assertNull(threadContext.getHeader("foo")); + assertEquals(nonEmptyContext ? "value" : null, threadContext.getHeader("not empty")); + + actionListener.onResponse(null); + + assertNull(threadContext.getHeader("foo")); + assertEquals(nonEmptyContext ? "value" : null, threadContext.getHeader("not empty")); } public void testOriginalContextIsPreservedAfterOnFailure() throws Exception { - try (ThreadContext threadContext = new ThreadContext(Settings.EMPTY)) { - final boolean nonEmptyContext = randomBoolean(); - if (nonEmptyContext) { - threadContext.putHeader("not empty", "value"); - } - final ContextPreservingActionListener actionListener; - try (ThreadContext.StoredContext ignore = threadContext.stashContext()) { - threadContext.putHeader("foo", "bar"); - final ActionListener delegate = new ActionListener() { - @Override - public void onResponse(Void aVoid) { - throw new RuntimeException("onResponse shouldn't be called"); - } - - @Override - public void onFailure(Exception e) { - assertEquals("bar", threadContext.getHeader("foo")); - assertNull(threadContext.getHeader("not empty")); - } - }; - - if (randomBoolean()) { - actionListener = new ContextPreservingActionListener<>(threadContext.newRestorableContext(true), delegate); - } else { - actionListener = ContextPreservingActionListener.wrapPreservingContext(delegate, threadContext); + ThreadContext threadContext = new ThreadContext(Settings.EMPTY); + final boolean nonEmptyContext = randomBoolean(); + if (nonEmptyContext) { + threadContext.putHeader("not empty", "value"); + } + final ContextPreservingActionListener actionListener; + try (ThreadContext.StoredContext ignore = threadContext.stashContext()) { + threadContext.putHeader("foo", "bar"); + final ActionListener delegate = new ActionListener() { + @Override + public void onResponse(Void aVoid) { + throw new RuntimeException("onResponse shouldn't be called"); } + @Override + public void onFailure(Exception e) { + assertEquals("bar", threadContext.getHeader("foo")); + assertNull(threadContext.getHeader("not empty")); + } + }; + + if (randomBoolean()) { + actionListener = new ContextPreservingActionListener<>(threadContext.newRestorableContext(true), delegate); + } else { + actionListener = ContextPreservingActionListener.wrapPreservingContext(delegate, threadContext); } - assertNull(threadContext.getHeader("foo")); - assertEquals(nonEmptyContext ? "value" : null, threadContext.getHeader("not empty")); - - actionListener.onFailure(null); - - assertNull(threadContext.getHeader("foo")); - assertEquals(nonEmptyContext ? "value" : null, threadContext.getHeader("not empty")); } - } + + assertNull(threadContext.getHeader("foo")); + assertEquals(nonEmptyContext ? "value" : null, threadContext.getHeader("not empty")); + + actionListener.onFailure(null); + + assertNull(threadContext.getHeader("foo")); + assertEquals(nonEmptyContext ? "value" : null, threadContext.getHeader("not empty")); +} public void testOriginalContextIsWhenListenerThrows() throws Exception { - try (ThreadContext threadContext = new ThreadContext(Settings.EMPTY)) { - final boolean nonEmptyContext = randomBoolean(); - if (nonEmptyContext) { - threadContext.putHeader("not empty", "value"); - } - final ContextPreservingActionListener actionListener; - try (ThreadContext.StoredContext ignore = threadContext.stashContext()) { - threadContext.putHeader("foo", "bar"); - final ActionListener delegate = new ActionListener() { - @Override - public void onResponse(Void aVoid) { - assertEquals("bar", threadContext.getHeader("foo")); - assertNull(threadContext.getHeader("not empty")); - throw new RuntimeException("onResponse called"); - } - - @Override - public void onFailure(Exception e) { - assertEquals("bar", threadContext.getHeader("foo")); - assertNull(threadContext.getHeader("not empty")); - throw new RuntimeException("onFailure called"); - } - }; - - if (randomBoolean()) { - actionListener = new ContextPreservingActionListener<>(threadContext.newRestorableContext(true), delegate); - } else { - actionListener = ContextPreservingActionListener.wrapPreservingContext(delegate, threadContext); - } - } - - assertNull(threadContext.getHeader("foo")); - assertEquals(nonEmptyContext ? "value" : null, threadContext.getHeader("not empty")); - - RuntimeException e = expectThrows(RuntimeException.class, () -> actionListener.onResponse(null)); - assertEquals("onResponse called", e.getMessage()); - - assertNull(threadContext.getHeader("foo")); - assertEquals(nonEmptyContext ? "value" : null, threadContext.getHeader("not empty")); - - e = expectThrows(RuntimeException.class, () -> actionListener.onFailure(null)); - assertEquals("onFailure called", e.getMessage()); - - assertNull(threadContext.getHeader("foo")); - assertEquals(nonEmptyContext ? "value" : null, threadContext.getHeader("not empty")); + ThreadContext threadContext = new ThreadContext(Settings.EMPTY); + final boolean nonEmptyContext = randomBoolean(); + if (nonEmptyContext) { + threadContext.putHeader("not empty", "value"); } + final ContextPreservingActionListener actionListener; + try (ThreadContext.StoredContext ignore = threadContext.stashContext()) { + threadContext.putHeader("foo", "bar"); + final ActionListener delegate = new ActionListener() { + @Override + public void onResponse(Void aVoid) { + assertEquals("bar", threadContext.getHeader("foo")); + assertNull(threadContext.getHeader("not empty")); + throw new RuntimeException("onResponse called"); + } + + @Override + public void onFailure(Exception e) { + assertEquals("bar", threadContext.getHeader("foo")); + assertNull(threadContext.getHeader("not empty")); + throw new RuntimeException("onFailure called"); + } + }; + + if (randomBoolean()) { + actionListener = new ContextPreservingActionListener<>(threadContext.newRestorableContext(true), delegate); + } else { + actionListener = ContextPreservingActionListener.wrapPreservingContext(delegate, threadContext); + } + } + + assertNull(threadContext.getHeader("foo")); + assertEquals(nonEmptyContext ? "value" : null, threadContext.getHeader("not empty")); + + RuntimeException e = expectThrows(RuntimeException.class, () -> actionListener.onResponse(null)); + assertEquals("onResponse called", e.getMessage()); + + assertNull(threadContext.getHeader("foo")); + assertEquals(nonEmptyContext ? "value" : null, threadContext.getHeader("not empty")); + + e = expectThrows(RuntimeException.class, () -> actionListener.onFailure(null)); + assertEquals("onFailure called", e.getMessage()); + + assertNull(threadContext.getHeader("foo")); + assertEquals(nonEmptyContext ? "value" : null, threadContext.getHeader("not empty")); } } diff --git a/server/src/test/java/org/elasticsearch/common/logging/DeprecationLoggerTests.java b/server/src/test/java/org/elasticsearch/common/logging/DeprecationLoggerTests.java index 71f3e9b03056..e7a78454e57f 100644 --- a/server/src/test/java/org/elasticsearch/common/logging/DeprecationLoggerTests.java +++ b/server/src/test/java/org/elasticsearch/common/logging/DeprecationLoggerTests.java @@ -42,7 +42,6 @@ import java.security.Permissions; import java.security.PrivilegedAction; import java.security.ProtectionDomain; import java.util.Collections; -import java.util.HashSet; import java.util.List; import java.util.Locale; import java.util.Map; @@ -78,136 +77,118 @@ public class DeprecationLoggerTests extends ESTestCase { } public void testAddsHeaderWithThreadContext() throws IOException { - try (ThreadContext threadContext = new ThreadContext(Settings.EMPTY)) { - final Set threadContexts = Collections.singleton(threadContext); + ThreadContext threadContext = new ThreadContext(Settings.EMPTY); + final Set threadContexts = Collections.singleton(threadContext); - final String param = randomAlphaOfLengthBetween(1, 5); - logger.deprecated(threadContexts, "A simple message [{}]", param); + final String param = randomAlphaOfLengthBetween(1, 5); + logger.deprecated(threadContexts, "A simple message [{}]", param); - final Map> responseHeaders = threadContext.getResponseHeaders(); + final Map> responseHeaders = threadContext.getResponseHeaders(); - assertThat(responseHeaders.size(), equalTo(1)); - final List responses = responseHeaders.get("Warning"); - assertThat(responses, hasSize(1)); - assertThat(responses.get(0), warningValueMatcher); - assertThat(responses.get(0), containsString("\"A simple message [" + param + "]\"")); - } + assertThat(responseHeaders.size(), equalTo(1)); + final List responses = responseHeaders.get("Warning"); + assertThat(responses, hasSize(1)); + assertThat(responses.get(0), warningValueMatcher); + assertThat(responses.get(0), containsString("\"A simple message [" + param + "]\"")); } public void testContainingNewline() throws IOException { - try (ThreadContext threadContext = new ThreadContext(Settings.EMPTY)) { - final Set threadContexts = Collections.singleton(threadContext); + ThreadContext threadContext = new ThreadContext(Settings.EMPTY); + final Set threadContexts = Collections.singleton(threadContext); - logger.deprecated(threadContexts, "this message contains a newline\n"); + logger.deprecated(threadContexts, "this message contains a newline\n"); - final Map> responseHeaders = threadContext.getResponseHeaders(); + final Map> responseHeaders = threadContext.getResponseHeaders(); - assertThat(responseHeaders.size(), equalTo(1)); - final List responses = responseHeaders.get("Warning"); - assertThat(responses, hasSize(1)); - assertThat(responses.get(0), warningValueMatcher); - assertThat(responses.get(0), containsString("\"this message contains a newline%0A\"")); - } + assertThat(responseHeaders.size(), equalTo(1)); + final List responses = responseHeaders.get("Warning"); + assertThat(responses, hasSize(1)); + assertThat(responses.get(0), warningValueMatcher); + assertThat(responses.get(0), containsString("\"this message contains a newline%0A\"")); } public void testSurrogatePair() throws IOException { - try (ThreadContext threadContext = new ThreadContext(Settings.EMPTY)) { - final Set threadContexts = Collections.singleton(threadContext); + ThreadContext threadContext = new ThreadContext(Settings.EMPTY); + final Set threadContexts = Collections.singleton(threadContext); - logger.deprecated(threadContexts, "this message contains a surrogate pair 😱"); + logger.deprecated(threadContexts, "this message contains a surrogate pair 😱"); - final Map> responseHeaders = threadContext.getResponseHeaders(); + final Map> responseHeaders = threadContext.getResponseHeaders(); - assertThat(responseHeaders.size(), equalTo(1)); - final List responses = responseHeaders.get("Warning"); - assertThat(responses, hasSize(1)); - assertThat(responses.get(0), warningValueMatcher); + assertThat(responseHeaders.size(), equalTo(1)); + final List responses = responseHeaders.get("Warning"); + assertThat(responses, hasSize(1)); + assertThat(responses.get(0), warningValueMatcher); - // convert UTF-16 to UTF-8 by hand to show the hard-coded constant below is correct - assertThat("😱", equalTo("\uD83D\uDE31")); - final int code = 0x10000 + ((0xD83D & 0x3FF) << 10) + (0xDE31 & 0x3FF); - @SuppressWarnings("PointlessBitwiseExpression") - final int[] points = new int[] { - (code >> 18) & 0x07 | 0xF0, - (code >> 12) & 0x3F | 0x80, - (code >> 6) & 0x3F | 0x80, - (code >> 0) & 0x3F | 0x80}; - final StringBuilder sb = new StringBuilder(); - // noinspection ForLoopReplaceableByForEach - for (int i = 0; i < points.length; i++) { - sb.append("%").append(Integer.toString(points[i], 16).toUpperCase(Locale.ROOT)); - } - assertThat(sb.toString(), equalTo("%F0%9F%98%B1")); - assertThat(responses.get(0), containsString("\"this message contains a surrogate pair %F0%9F%98%B1\"")); + // convert UTF-16 to UTF-8 by hand to show the hard-coded constant below is correct + assertThat("😱", equalTo("\uD83D\uDE31")); + final int code = 0x10000 + ((0xD83D & 0x3FF) << 10) + (0xDE31 & 0x3FF); + @SuppressWarnings("PointlessBitwiseExpression") + final int[] points = new int[] { + (code >> 18) & 0x07 | 0xF0, + (code >> 12) & 0x3F | 0x80, + (code >> 6) & 0x3F | 0x80, + (code >> 0) & 0x3F | 0x80}; + final StringBuilder sb = new StringBuilder(); + // noinspection ForLoopReplaceableByForEach + for (int i = 0; i < points.length; i++) { + sb.append("%").append(Integer.toString(points[i], 16).toUpperCase(Locale.ROOT)); } + assertThat(sb.toString(), equalTo("%F0%9F%98%B1")); + assertThat(responses.get(0), containsString("\"this message contains a surrogate pair %F0%9F%98%B1\"")); } public void testAddsCombinedHeaderWithThreadContext() throws IOException { - try (ThreadContext threadContext = new ThreadContext(Settings.EMPTY)) { - final Set threadContexts = Collections.singleton(threadContext); + ThreadContext threadContext = new ThreadContext(Settings.EMPTY); + final Set threadContexts = Collections.singleton(threadContext); - final String param = randomAlphaOfLengthBetween(1, 5); - logger.deprecated(threadContexts, "A simple message [{}]", param); - final String second = randomAlphaOfLengthBetween(1, 10); - logger.deprecated(threadContexts, second); + final String param = randomAlphaOfLengthBetween(1, 5); + logger.deprecated(threadContexts, "A simple message [{}]", param); + final String second = randomAlphaOfLengthBetween(1, 10); + logger.deprecated(threadContexts, second); - final Map> responseHeaders = threadContext.getResponseHeaders(); + final Map> responseHeaders = threadContext.getResponseHeaders(); - assertEquals(1, responseHeaders.size()); + assertEquals(1, responseHeaders.size()); - final List responses = responseHeaders.get("Warning"); + final List responses = responseHeaders.get("Warning"); - assertEquals(2, responses.size()); - assertThat(responses.get(0), warningValueMatcher); - assertThat(responses.get(0), containsString("\"A simple message [" + param + "]\"")); - assertThat(responses.get(1), warningValueMatcher); - assertThat(responses.get(1), containsString("\"" + second + "\"")); - } + assertEquals(2, responses.size()); + assertThat(responses.get(0), warningValueMatcher); + assertThat(responses.get(0), containsString("\"A simple message [" + param + "]\"")); + assertThat(responses.get(1), warningValueMatcher); + assertThat(responses.get(1), containsString("\"" + second + "\"")); } public void testCanRemoveThreadContext() throws IOException { final String expected = "testCanRemoveThreadContext"; final String unexpected = "testCannotRemoveThreadContext"; - try (ThreadContext threadContext = new ThreadContext(Settings.EMPTY)) { - DeprecationLogger.setThreadContext(threadContext); - logger.deprecated(expected); - - { - final Map> responseHeaders = threadContext.getResponseHeaders(); - final List responses = responseHeaders.get("Warning"); - - assertThat(responses, hasSize(1)); - assertThat(responses.get(0), warningValueMatcher); - assertThat(responses.get(0), containsString(expected)); - } - - DeprecationLogger.removeThreadContext(threadContext); - logger.deprecated(unexpected); - - { - final Map> responseHeaders = threadContext.getResponseHeaders(); - final List responses = responseHeaders.get("Warning"); - - assertThat(responses, hasSize(1)); - assertThat(responses.get(0), warningValueMatcher); - assertThat(responses.get(0), containsString(expected)); - assertThat(responses.get(0), not(containsString(unexpected))); - } - } - } - - public void testIgnoresClosedThreadContext() throws IOException { ThreadContext threadContext = new ThreadContext(Settings.EMPTY); - Set threadContexts = new HashSet<>(1); + DeprecationLogger.setThreadContext(threadContext); + logger.deprecated(expected); - threadContexts.add(threadContext); + { + final Map> responseHeaders = threadContext.getResponseHeaders(); + final List responses = responseHeaders.get("Warning"); - threadContext.close(); + assertThat(responses, hasSize(1)); + assertThat(responses.get(0), warningValueMatcher); + assertThat(responses.get(0), containsString(expected)); + } - logger.deprecated(threadContexts, "Ignored logger message"); + DeprecationLogger.removeThreadContext(threadContext); + logger.deprecated(unexpected); - assertTrue(threadContexts.contains(threadContext)); + { + final Map> responseHeaders = threadContext.getResponseHeaders(); + final List responses = responseHeaders.get("Warning"); + + assertThat(responses, hasSize(1)); + assertThat(responses.get(0), warningValueMatcher); + assertThat(responses.get(0), containsString(expected)); + assertThat(responses.get(0), not(containsString(unexpected))); + } } public void testSafeWithoutThreadContext() { @@ -219,22 +200,20 @@ public class DeprecationLoggerTests extends ESTestCase { } public void testFailsWhenDoubleSettingSameThreadContext() throws IOException { - try (ThreadContext threadContext = new ThreadContext(Settings.EMPTY)) { - DeprecationLogger.setThreadContext(threadContext); + ThreadContext threadContext = new ThreadContext(Settings.EMPTY); + DeprecationLogger.setThreadContext(threadContext); - try { - expectThrows(IllegalStateException.class, () -> DeprecationLogger.setThreadContext(threadContext)); - } finally { - // cleanup after ourselves - DeprecationLogger.removeThreadContext(threadContext); - } + try { + expectThrows(IllegalStateException.class, () -> DeprecationLogger.setThreadContext(threadContext)); + } finally { + // cleanup after ourselves + DeprecationLogger.removeThreadContext(threadContext); } } public void testFailsWhenRemovingUnknownThreadContext() throws IOException { - try (ThreadContext threadContext = new ThreadContext(Settings.EMPTY)) { - expectThrows(IllegalStateException.class, () -> DeprecationLogger.removeThreadContext(threadContext)); - } + ThreadContext threadContext = new ThreadContext(Settings.EMPTY); + expectThrows(IllegalStateException.class, () -> DeprecationLogger.removeThreadContext(threadContext)); } public void testWarningValueFromWarningHeader() throws InterruptedException { @@ -274,21 +253,20 @@ public class DeprecationLoggerTests extends ESTestCase { Settings settings = Settings.builder() .put("http.max_warning_header_count", maxWarningHeaderCount) .build(); - try (ThreadContext threadContext = new ThreadContext(settings)) { - final Set threadContexts = Collections.singleton(threadContext); - // try to log three warning messages - logger.deprecated(threadContexts, "A simple message 1"); - logger.deprecated(threadContexts, "A simple message 2"); - logger.deprecated(threadContexts, "A simple message 3"); - final Map> responseHeaders = threadContext.getResponseHeaders(); - final List responses = responseHeaders.get("Warning"); + ThreadContext threadContext = new ThreadContext(settings); + final Set threadContexts = Collections.singleton(threadContext); + // try to log three warning messages + logger.deprecated(threadContexts, "A simple message 1"); + logger.deprecated(threadContexts, "A simple message 2"); + logger.deprecated(threadContexts, "A simple message 3"); + final Map> responseHeaders = threadContext.getResponseHeaders(); + final List responses = responseHeaders.get("Warning"); - assertEquals(maxWarningHeaderCount, responses.size()); - assertThat(responses.get(0), warningValueMatcher); - assertThat(responses.get(0), containsString("\"A simple message 1")); - assertThat(responses.get(1), warningValueMatcher); - assertThat(responses.get(1), containsString("\"A simple message 2")); - } + assertEquals(maxWarningHeaderCount, responses.size()); + assertThat(responses.get(0), warningValueMatcher); + assertThat(responses.get(0), containsString("\"A simple message 1")); + assertThat(responses.get(1), warningValueMatcher); + assertThat(responses.get(1), containsString("\"A simple message 2")); } public void testWarningHeaderSizeSetting() throws IOException{ @@ -302,23 +280,22 @@ public class DeprecationLoggerTests extends ESTestCase { String message2 = new String(arr, StandardCharsets.UTF_8) + "2"; String message3 = new String(arr, StandardCharsets.UTF_8) + "3"; - try (ThreadContext threadContext = new ThreadContext(settings)) { - final Set threadContexts = Collections.singleton(threadContext); - // try to log three warning messages - logger.deprecated(threadContexts, message1); - logger.deprecated(threadContexts, message2); - logger.deprecated(threadContexts, message3); - final Map> responseHeaders = threadContext.getResponseHeaders(); - final List responses = responseHeaders.get("Warning"); + ThreadContext threadContext = new ThreadContext(settings); + final Set threadContexts = Collections.singleton(threadContext); + // try to log three warning messages + logger.deprecated(threadContexts, message1); + logger.deprecated(threadContexts, message2); + logger.deprecated(threadContexts, message3); + final Map> responseHeaders = threadContext.getResponseHeaders(); + final List responses = responseHeaders.get("Warning"); - long warningHeadersSize = 0L; - for (String response : responses){ - warningHeadersSize += "Warning".getBytes(StandardCharsets.UTF_8).length + - response.getBytes(StandardCharsets.UTF_8).length; - } - // assert that the size of all warning headers is less or equal to 1Kb - assertTrue(warningHeadersSize <= 1024); + long warningHeadersSize = 0L; + for (String response : responses){ + warningHeadersSize += "Warning".getBytes(StandardCharsets.UTF_8).length + + response.getBytes(StandardCharsets.UTF_8).length; } + // assert that the size of all warning headers is less or equal to 1Kb + assertTrue(warningHeadersSize <= 1024); } @SuppressLoggerChecks(reason = "Safe as this is using mockito") public void testLogPermissions() { diff --git a/server/src/test/java/org/elasticsearch/common/util/concurrent/AsyncIOProcessorTests.java b/server/src/test/java/org/elasticsearch/common/util/concurrent/AsyncIOProcessorTests.java index 56aa797caa6e..e155b826269b 100644 --- a/server/src/test/java/org/elasticsearch/common/util/concurrent/AsyncIOProcessorTests.java +++ b/server/src/test/java/org/elasticsearch/common/util/concurrent/AsyncIOProcessorTests.java @@ -21,7 +21,6 @@ package org.elasticsearch.common.util.concurrent; import org.elasticsearch.common.collect.Tuple; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.test.ESTestCase; -import org.junit.After; import org.junit.Before; import java.io.IOException; @@ -47,11 +46,6 @@ public class AsyncIOProcessorTests extends ESTestCase { threadContext = new ThreadContext(Settings.EMPTY); } - @After - public void tearDownThreadContext() { - threadContext.close(); - } - public void testPut() throws InterruptedException { boolean blockInternal = randomBoolean(); AtomicInteger received = new AtomicInteger(0); diff --git a/server/src/test/java/org/elasticsearch/common/util/concurrent/QueueResizingEsThreadPoolExecutorTests.java b/server/src/test/java/org/elasticsearch/common/util/concurrent/QueueResizingEsThreadPoolExecutorTests.java index 7ffe6a4090a2..e08a6ca44331 100644 --- a/server/src/test/java/org/elasticsearch/common/util/concurrent/QueueResizingEsThreadPoolExecutorTests.java +++ b/server/src/test/java/org/elasticsearch/common/util/concurrent/QueueResizingEsThreadPoolExecutorTests.java @@ -64,7 +64,6 @@ public class QueueResizingEsThreadPoolExecutorTests extends ESTestCase { }); executor.shutdown(); executor.awaitTermination(10, TimeUnit.SECONDS); - context.close(); } public void testAutoQueueSizingUp() throws Exception { @@ -93,7 +92,6 @@ public class QueueResizingEsThreadPoolExecutorTests extends ESTestCase { }); executor.shutdown(); executor.awaitTermination(10, TimeUnit.SECONDS); - context.close(); } public void testAutoQueueSizingDown() throws Exception { @@ -121,7 +119,6 @@ public class QueueResizingEsThreadPoolExecutorTests extends ESTestCase { }); executor.shutdown(); executor.awaitTermination(10, TimeUnit.SECONDS); - context.close(); } public void testAutoQueueSizingWithMin() throws Exception { @@ -151,7 +148,6 @@ public class QueueResizingEsThreadPoolExecutorTests extends ESTestCase { }); executor.shutdown(); executor.awaitTermination(10, TimeUnit.SECONDS); - context.close(); } public void testAutoQueueSizingWithMax() throws Exception { @@ -181,7 +177,6 @@ public class QueueResizingEsThreadPoolExecutorTests extends ESTestCase { }); executor.shutdown(); executor.awaitTermination(10, TimeUnit.SECONDS); - context.close(); } public void testExecutionEWMACalculation() throws Exception { @@ -222,7 +217,6 @@ public class QueueResizingEsThreadPoolExecutorTests extends ESTestCase { executor.shutdown(); executor.awaitTermination(10, TimeUnit.SECONDS); - context.close(); } /** Use a runnable wrapper that simulates a task with unknown failures. */ @@ -244,7 +238,6 @@ public class QueueResizingEsThreadPoolExecutorTests extends ESTestCase { executeTask(executor, 1); executor.shutdown(); executor.awaitTermination(10, TimeUnit.SECONDS); - context.close(); } private Function fastWrapper() { diff --git a/server/src/test/java/org/elasticsearch/common/util/concurrent/ThreadContextTests.java b/server/src/test/java/org/elasticsearch/common/util/concurrent/ThreadContextTests.java index 9729aca29418..46c0d6a58992 100644 --- a/server/src/test/java/org/elasticsearch/common/util/concurrent/ThreadContextTests.java +++ b/server/src/test/java/org/elasticsearch/common/util/concurrent/ThreadContextTests.java @@ -244,35 +244,6 @@ public class ThreadContextTests extends ESTestCase { assertEquals("bar", threadContext.getHeader("foo")); } - public void testAccessClosed() throws IOException { - Settings build = Settings.builder().put("request.headers.default", "1").build(); - ThreadContext threadContext = new ThreadContext(build); - threadContext.putHeader("foo", "bar"); - threadContext.putTransient("ctx.foo", 1); - - threadContext.close(); - try { - threadContext.getHeader("foo"); - fail(); - } catch (IllegalStateException ise) { - assertEquals("threadcontext is already closed", ise.getMessage()); - } - - try { - threadContext.putTransient("foo", new Object()); - fail(); - } catch (IllegalStateException ise) { - assertEquals("threadcontext is already closed", ise.getMessage()); - } - - try { - threadContext.putHeader("boom", "boom"); - fail(); - } catch (IllegalStateException ise) { - assertEquals("threadcontext is already closed", ise.getMessage()); - } - } - public void testSerialize() throws IOException { Settings build = Settings.builder().put("request.headers.default", "1").build(); ThreadContext threadContext = new ThreadContext(build); @@ -397,244 +368,238 @@ public class ThreadContextTests extends ESTestCase { } public void testPreserveContext() throws IOException { - try (ThreadContext threadContext = new ThreadContext(Settings.EMPTY)) { - Runnable withContext; + ThreadContext threadContext = new ThreadContext(Settings.EMPTY); + Runnable withContext; - // Create a runnable that should run with some header - try (ThreadContext.StoredContext ignored = threadContext.stashContext()) { - threadContext.putHeader("foo", "bar"); - withContext = threadContext.preserveContext(sometimesAbstractRunnable(() -> { - assertEquals("bar", threadContext.getHeader("foo")); - })); - } - - // We don't see the header outside of the runnable - assertNull(threadContext.getHeader("foo")); - - // But we do inside of it - withContext.run(); - - // but not after - assertNull(threadContext.getHeader("foo")); + // Create a runnable that should run with some header + try (ThreadContext.StoredContext ignored = threadContext.stashContext()) { + threadContext.putHeader("foo", "bar"); + withContext = threadContext.preserveContext(sometimesAbstractRunnable(() -> { + assertEquals("bar", threadContext.getHeader("foo")); + })); } + + // We don't see the header outside of the runnable + assertNull(threadContext.getHeader("foo")); + + // But we do inside of it + withContext.run(); + + // but not after + assertNull(threadContext.getHeader("foo")); } public void testPreserveContextKeepsOriginalContextWhenCalledTwice() throws IOException { - try (ThreadContext threadContext = new ThreadContext(Settings.EMPTY)) { - Runnable originalWithContext; - Runnable withContext; + ThreadContext threadContext = new ThreadContext(Settings.EMPTY); + Runnable originalWithContext; + Runnable withContext; - // Create a runnable that should run with some header - try (ThreadContext.StoredContext ignored = threadContext.stashContext()) { - threadContext.putHeader("foo", "bar"); - withContext = threadContext.preserveContext(sometimesAbstractRunnable(() -> { - assertEquals("bar", threadContext.getHeader("foo")); - })); - } - - // Now attempt to rewrap it - originalWithContext = withContext; - try (ThreadContext.StoredContext ignored = threadContext.stashContext()) { - threadContext.putHeader("foo", "zot"); - withContext = threadContext.preserveContext(withContext); - } - - // We get the original context inside the runnable - withContext.run(); - - // In fact the second wrapping didn't even change it - assertThat(withContext, sameInstance(originalWithContext)); + // Create a runnable that should run with some header + try (ThreadContext.StoredContext ignored = threadContext.stashContext()) { + threadContext.putHeader("foo", "bar"); + withContext = threadContext.preserveContext(sometimesAbstractRunnable(() -> { + assertEquals("bar", threadContext.getHeader("foo")); + })); } + + // Now attempt to rewrap it + originalWithContext = withContext; + try (ThreadContext.StoredContext ignored = threadContext.stashContext()) { + threadContext.putHeader("foo", "zot"); + withContext = threadContext.preserveContext(withContext); + } + + // We get the original context inside the runnable + withContext.run(); + + // In fact the second wrapping didn't even change it + assertThat(withContext, sameInstance(originalWithContext)); } public void testPreservesThreadsOriginalContextOnRunException() throws IOException { - try (ThreadContext threadContext = new ThreadContext(Settings.EMPTY)) { - Runnable withContext; + ThreadContext threadContext = new ThreadContext(Settings.EMPTY); + Runnable withContext; - // create a abstract runnable, add headers and transient objects and verify in the methods - try (ThreadContext.StoredContext ignored = threadContext.stashContext()) { - threadContext.putHeader("foo", "bar"); - boolean systemContext = randomBoolean(); - if (systemContext) { - threadContext.markAsSystemContext(); - } - threadContext.putTransient("foo", "bar_transient"); - withContext = threadContext.preserveContext(new AbstractRunnable() { - - @Override - public void onAfter() { - assertEquals(systemContext, threadContext.isSystemContext()); - assertEquals("bar", threadContext.getHeader("foo")); - assertEquals("bar_transient", threadContext.getTransient("foo")); - assertNotNull(threadContext.getTransient("failure")); - assertEquals("exception from doRun", ((RuntimeException) threadContext.getTransient("failure")).getMessage()); - assertFalse(threadContext.isDefaultContext()); - threadContext.putTransient("after", "after"); - } - - @Override - public void onFailure(Exception e) { - assertEquals(systemContext, threadContext.isSystemContext()); - assertEquals("exception from doRun", e.getMessage()); - assertEquals("bar", threadContext.getHeader("foo")); - assertEquals("bar_transient", threadContext.getTransient("foo")); - assertFalse(threadContext.isDefaultContext()); - threadContext.putTransient("failure", e); - } - - @Override - protected void doRun() throws Exception { - assertEquals(systemContext, threadContext.isSystemContext()); - assertEquals("bar", threadContext.getHeader("foo")); - assertEquals("bar_transient", threadContext.getTransient("foo")); - assertFalse(threadContext.isDefaultContext()); - throw new RuntimeException("exception from doRun"); - } - }); + // create a abstract runnable, add headers and transient objects and verify in the methods + try (ThreadContext.StoredContext ignored = threadContext.stashContext()) { + threadContext.putHeader("foo", "bar"); + boolean systemContext = randomBoolean(); + if (systemContext) { + threadContext.markAsSystemContext(); } + threadContext.putTransient("foo", "bar_transient"); + withContext = threadContext.preserveContext(new AbstractRunnable() { - // We don't see the header outside of the runnable - assertNull(threadContext.getHeader("foo")); - assertNull(threadContext.getTransient("foo")); - assertNull(threadContext.getTransient("failure")); - assertNull(threadContext.getTransient("after")); - assertTrue(threadContext.isDefaultContext()); + @Override + public void onAfter() { + assertEquals(systemContext, threadContext.isSystemContext()); + assertEquals("bar", threadContext.getHeader("foo")); + assertEquals("bar_transient", threadContext.getTransient("foo")); + assertNotNull(threadContext.getTransient("failure")); + assertEquals("exception from doRun", ((RuntimeException) threadContext.getTransient("failure")).getMessage()); + assertFalse(threadContext.isDefaultContext()); + threadContext.putTransient("after", "after"); + } - // But we do inside of it - withContext.run(); - - // verify not seen after - assertNull(threadContext.getHeader("foo")); - assertNull(threadContext.getTransient("foo")); - assertNull(threadContext.getTransient("failure")); - assertNull(threadContext.getTransient("after")); - assertTrue(threadContext.isDefaultContext()); - - // repeat with regular runnable - try (ThreadContext.StoredContext ignored = threadContext.stashContext()) { - threadContext.putHeader("foo", "bar"); - threadContext.putTransient("foo", "bar_transient"); - withContext = threadContext.preserveContext(() -> { + @Override + public void onFailure(Exception e) { + assertEquals(systemContext, threadContext.isSystemContext()); + assertEquals("exception from doRun", e.getMessage()); assertEquals("bar", threadContext.getHeader("foo")); assertEquals("bar_transient", threadContext.getTransient("foo")); assertFalse(threadContext.isDefaultContext()); - threadContext.putTransient("run", true); - throw new RuntimeException("exception from run"); - }); - } + threadContext.putTransient("failure", e); + } - assertNull(threadContext.getHeader("foo")); - assertNull(threadContext.getTransient("foo")); - assertNull(threadContext.getTransient("run")); - assertTrue(threadContext.isDefaultContext()); - - final Runnable runnable = withContext; - RuntimeException e = expectThrows(RuntimeException.class, runnable::run); - assertEquals("exception from run", e.getMessage()); - assertNull(threadContext.getHeader("foo")); - assertNull(threadContext.getTransient("foo")); - assertNull(threadContext.getTransient("run")); - assertTrue(threadContext.isDefaultContext()); + @Override + protected void doRun() throws Exception { + assertEquals(systemContext, threadContext.isSystemContext()); + assertEquals("bar", threadContext.getHeader("foo")); + assertEquals("bar_transient", threadContext.getTransient("foo")); + assertFalse(threadContext.isDefaultContext()); + throw new RuntimeException("exception from doRun"); + } + }); } + + // We don't see the header outside of the runnable + assertNull(threadContext.getHeader("foo")); + assertNull(threadContext.getTransient("foo")); + assertNull(threadContext.getTransient("failure")); + assertNull(threadContext.getTransient("after")); + assertTrue(threadContext.isDefaultContext()); + + // But we do inside of it + withContext.run(); + + // verify not seen after + assertNull(threadContext.getHeader("foo")); + assertNull(threadContext.getTransient("foo")); + assertNull(threadContext.getTransient("failure")); + assertNull(threadContext.getTransient("after")); + assertTrue(threadContext.isDefaultContext()); + + // repeat with regular runnable + try (ThreadContext.StoredContext ignored = threadContext.stashContext()) { + threadContext.putHeader("foo", "bar"); + threadContext.putTransient("foo", "bar_transient"); + withContext = threadContext.preserveContext(() -> { + assertEquals("bar", threadContext.getHeader("foo")); + assertEquals("bar_transient", threadContext.getTransient("foo")); + assertFalse(threadContext.isDefaultContext()); + threadContext.putTransient("run", true); + throw new RuntimeException("exception from run"); + }); + } + + assertNull(threadContext.getHeader("foo")); + assertNull(threadContext.getTransient("foo")); + assertNull(threadContext.getTransient("run")); + assertTrue(threadContext.isDefaultContext()); + + final Runnable runnable = withContext; + RuntimeException e = expectThrows(RuntimeException.class, runnable::run); + assertEquals("exception from run", e.getMessage()); + assertNull(threadContext.getHeader("foo")); + assertNull(threadContext.getTransient("foo")); + assertNull(threadContext.getTransient("run")); + assertTrue(threadContext.isDefaultContext()); } public void testPreservesThreadsOriginalContextOnFailureException() throws IOException { - try (ThreadContext threadContext = new ThreadContext(Settings.EMPTY)) { - Runnable withContext; + ThreadContext threadContext = new ThreadContext(Settings.EMPTY); + Runnable withContext; - // a runnable that throws from onFailure - try (ThreadContext.StoredContext ignored = threadContext.stashContext()) { - threadContext.putHeader("foo", "bar"); - threadContext.putTransient("foo", "bar_transient"); - withContext = threadContext.preserveContext(new AbstractRunnable() { - @Override - public void onFailure(Exception e) { - throw new RuntimeException("from onFailure", e); - } + // a runnable that throws from onFailure + try (ThreadContext.StoredContext ignored = threadContext.stashContext()) { + threadContext.putHeader("foo", "bar"); + threadContext.putTransient("foo", "bar_transient"); + withContext = threadContext.preserveContext(new AbstractRunnable() { + @Override + public void onFailure(Exception e) { + throw new RuntimeException("from onFailure", e); + } - @Override - protected void doRun() throws Exception { - assertEquals("bar", threadContext.getHeader("foo")); - assertEquals("bar_transient", threadContext.getTransient("foo")); - assertFalse(threadContext.isDefaultContext()); - throw new RuntimeException("from doRun"); - } - }); - } - - // We don't see the header outside of the runnable - assertNull(threadContext.getHeader("foo")); - assertNull(threadContext.getTransient("foo")); - assertTrue(threadContext.isDefaultContext()); - - // But we do inside of it - RuntimeException e = expectThrows(RuntimeException.class, withContext::run); - assertEquals("from onFailure", e.getMessage()); - assertEquals("from doRun", e.getCause().getMessage()); - - // but not after - assertNull(threadContext.getHeader("foo")); - assertNull(threadContext.getTransient("foo")); - assertTrue(threadContext.isDefaultContext()); + @Override + protected void doRun() throws Exception { + assertEquals("bar", threadContext.getHeader("foo")); + assertEquals("bar_transient", threadContext.getTransient("foo")); + assertFalse(threadContext.isDefaultContext()); + throw new RuntimeException("from doRun"); + } + }); } + + // We don't see the header outside of the runnable + assertNull(threadContext.getHeader("foo")); + assertNull(threadContext.getTransient("foo")); + assertTrue(threadContext.isDefaultContext()); + + // But we do inside of it + RuntimeException e = expectThrows(RuntimeException.class, withContext::run); + assertEquals("from onFailure", e.getMessage()); + assertEquals("from doRun", e.getCause().getMessage()); + + // but not after + assertNull(threadContext.getHeader("foo")); + assertNull(threadContext.getTransient("foo")); + assertTrue(threadContext.isDefaultContext()); } public void testPreservesThreadsOriginalContextOnAfterException() throws IOException { - try (ThreadContext threadContext = new ThreadContext(Settings.EMPTY)) { - Runnable withContext; + ThreadContext threadContext = new ThreadContext(Settings.EMPTY); + Runnable withContext; - // a runnable that throws from onAfter - try (ThreadContext.StoredContext ignored = threadContext.stashContext()) { - threadContext.putHeader("foo", "bar"); - threadContext.putTransient("foo", "bar_transient"); - withContext = threadContext.preserveContext(new AbstractRunnable() { + // a runnable that throws from onAfter + try (ThreadContext.StoredContext ignored = threadContext.stashContext()) { + threadContext.putHeader("foo", "bar"); + threadContext.putTransient("foo", "bar_transient"); + withContext = threadContext.preserveContext(new AbstractRunnable() { - @Override - public void onAfter() { - throw new RuntimeException("from onAfter"); - } + @Override + public void onAfter() { + throw new RuntimeException("from onAfter"); + } - @Override - public void onFailure(Exception e) { - throw new RuntimeException("from onFailure", e); - } + @Override + public void onFailure(Exception e) { + throw new RuntimeException("from onFailure", e); + } - @Override - protected void doRun() throws Exception { - assertEquals("bar", threadContext.getHeader("foo")); - assertEquals("bar_transient", threadContext.getTransient("foo")); - assertFalse(threadContext.isDefaultContext()); - } - }); - } - - // We don't see the header outside of the runnable - assertNull(threadContext.getHeader("foo")); - assertNull(threadContext.getTransient("foo")); - assertTrue(threadContext.isDefaultContext()); - - // But we do inside of it - RuntimeException e = expectThrows(RuntimeException.class, withContext::run); - assertEquals("from onAfter", e.getMessage()); - assertNull(e.getCause()); - - // but not after - assertNull(threadContext.getHeader("foo")); - assertNull(threadContext.getTransient("foo")); - assertTrue(threadContext.isDefaultContext()); + @Override + protected void doRun() throws Exception { + assertEquals("bar", threadContext.getHeader("foo")); + assertEquals("bar_transient", threadContext.getTransient("foo")); + assertFalse(threadContext.isDefaultContext()); + } + }); } + + // We don't see the header outside of the runnable + assertNull(threadContext.getHeader("foo")); + assertNull(threadContext.getTransient("foo")); + assertTrue(threadContext.isDefaultContext()); + + // But we do inside of it + RuntimeException e = expectThrows(RuntimeException.class, withContext::run); + assertEquals("from onAfter", e.getMessage()); + assertNull(e.getCause()); + + // but not after + assertNull(threadContext.getHeader("foo")); + assertNull(threadContext.getTransient("foo")); + assertTrue(threadContext.isDefaultContext()); } public void testMarkAsSystemContext() throws IOException { - try (ThreadContext threadContext = new ThreadContext(Settings.EMPTY)) { - assertFalse(threadContext.isSystemContext()); - try (ThreadContext.StoredContext context = threadContext.stashContext()) { - assertFalse(threadContext.isSystemContext()); - threadContext.markAsSystemContext(); - assertTrue(threadContext.isSystemContext()); - } + ThreadContext threadContext = new ThreadContext(Settings.EMPTY); + assertFalse(threadContext.isSystemContext()); + try (ThreadContext.StoredContext context = threadContext.stashContext()) { assertFalse(threadContext.isSystemContext()); + threadContext.markAsSystemContext(); + assertTrue(threadContext.isSystemContext()); } + assertFalse(threadContext.isSystemContext()); } public void testPutHeaders() { diff --git a/server/src/test/java/org/elasticsearch/node/NodeTests.java b/server/src/test/java/org/elasticsearch/node/NodeTests.java index 5e06e557975e..6f6b89e31f1b 100644 --- a/server/src/test/java/org/elasticsearch/node/NodeTests.java +++ b/server/src/test/java/org/elasticsearch/node/NodeTests.java @@ -166,7 +166,6 @@ public class NodeTests extends ESTestCase { assertTrue(node.awaitClose(1, TimeUnit.DAYS)); } - @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/42577") public void testCloseRaceWithTaskExecution() throws Exception { Node node = new MockNode(baseSettings().build(), basePlugins()); node.start(); diff --git a/server/src/test/java/org/elasticsearch/threadpool/ThreadPoolTests.java b/server/src/test/java/org/elasticsearch/threadpool/ThreadPoolTests.java index 0be9858269af..68aff95419ee 100644 --- a/server/src/test/java/org/elasticsearch/threadpool/ThreadPoolTests.java +++ b/server/src/test/java/org/elasticsearch/threadpool/ThreadPoolTests.java @@ -62,8 +62,7 @@ public class ThreadPoolTests extends ESTestCase { // the delta can be large, we just care it is the same order of magnitude assertTrue("thread pool cached absolute time " + gotTime + " is too far from real current time " + currentTime, delta < 10000); } finally { - threadPool.shutdown(); - threadPool.close(); + terminate(threadPool); } } diff --git a/server/src/test/java/org/elasticsearch/transport/TransportLoggerTests.java b/server/src/test/java/org/elasticsearch/transport/TransportLoggerTests.java index a5f6354b6fcd..1d62e3d9ab66 100644 --- a/server/src/test/java/org/elasticsearch/transport/TransportLoggerTests.java +++ b/server/src/test/java/org/elasticsearch/transport/TransportLoggerTests.java @@ -90,9 +90,8 @@ public class TransportLoggerTests extends ESTestCase { private BytesReference buildRequest() throws IOException { try (BytesStreamOutput messageOutput = new BytesStreamOutput()) { messageOutput.setVersion(Version.CURRENT); - try (ThreadContext context = new ThreadContext(Settings.EMPTY)) { - context.writeTo(messageOutput); - } + ThreadContext context = new ThreadContext(Settings.EMPTY); + context.writeTo(messageOutput); messageOutput.writeString(ClusterStatsAction.NAME); new ClusterStatsRequest().writeTo(messageOutput); BytesReference messageBody = messageOutput.bytes(); diff --git a/test/framework/src/main/java/org/elasticsearch/test/ESTestCase.java b/test/framework/src/main/java/org/elasticsearch/test/ESTestCase.java index e538790f871b..4d1848e5b3a9 100644 --- a/test/framework/src/main/java/org/elasticsearch/test/ESTestCase.java +++ b/test/framework/src/main/java/org/elasticsearch/test/ESTestCase.java @@ -370,7 +370,8 @@ public abstract class ESTestCase extends LuceneTestCase { // initialized if (threadContext != null) { ensureNoWarnings(); - assert threadContext == null; + DeprecationLogger.removeThreadContext(threadContext); + threadContext = null; } ensureAllSearchContextsReleased(); ensureCheckIndexPassed(); @@ -393,7 +394,7 @@ public abstract class ESTestCase extends LuceneTestCase { final List warnings = threadContext.getResponseHeaders().get("Warning"); assertNull("unexpected warning headers", warnings); } finally { - resetDeprecationLogger(false); + resetDeprecationLogger(); } } @@ -434,26 +435,16 @@ public abstract class ESTestCase extends LuceneTestCase { + Arrays.asList(expectedWarnings) + "\nActual: " + actualWarnings, expectedWarnings.length, actualWarnings.size()); } finally { - resetDeprecationLogger(true); + resetDeprecationLogger(); } } /** - * Reset the deprecation logger by removing the current thread context, and setting a new thread context if {@code setNewThreadContext} - * is set to {@code true} and otherwise clearing the current thread context. - * - * @param setNewThreadContext whether or not to attach a new thread context to the deprecation logger + * Reset the deprecation logger by clearing the current thread context. */ - private void resetDeprecationLogger(final boolean setNewThreadContext) { - // "clear" current warning headers by setting a new ThreadContext - DeprecationLogger.removeThreadContext(this.threadContext); - this.threadContext.close(); - if (setNewThreadContext) { - this.threadContext = new ThreadContext(Settings.EMPTY); - DeprecationLogger.setThreadContext(this.threadContext); - } else { - this.threadContext = null; - } + private void resetDeprecationLogger() { + // "clear" context by stashing current values and dropping the returned StoredContext + threadContext.stashContext(); } private static final List statusData = new ArrayList<>(); diff --git a/test/framework/src/main/java/org/elasticsearch/test/rest/ESRestTestCase.java b/test/framework/src/main/java/org/elasticsearch/test/rest/ESRestTestCase.java index 625c129767d5..6349c2f10548 100644 --- a/test/framework/src/main/java/org/elasticsearch/test/rest/ESRestTestCase.java +++ b/test/framework/src/main/java/org/elasticsearch/test/rest/ESRestTestCase.java @@ -881,14 +881,13 @@ public abstract class ESRestTestCase extends ESTestCase { throw new RuntimeException("Error setting up ssl", e); } } - try (ThreadContext threadContext = new ThreadContext(settings)) { - Header[] defaultHeaders = new Header[threadContext.getHeaders().size()]; - int i = 0; - for (Map.Entry entry : threadContext.getHeaders().entrySet()) { - defaultHeaders[i++] = new BasicHeader(entry.getKey(), entry.getValue()); - } - builder.setDefaultHeaders(defaultHeaders); + Map headers = ThreadContext.buildDefaultHeaders(settings); + Header[] defaultHeaders = new Header[headers.size()]; + int i = 0; + for (Map.Entry entry : headers.entrySet()) { + defaultHeaders[i++] = new BasicHeader(entry.getKey(), entry.getValue()); } + builder.setDefaultHeaders(defaultHeaders); final String socketTimeoutString = Objects.requireNonNullElse(settings.get(CLIENT_SOCKET_TIMEOUT), "60s"); final TimeValue socketTimeout = TimeValue.parseTimeValue(socketTimeoutString, CLIENT_SOCKET_TIMEOUT); builder.setRequestConfigCallback(conf -> conf.setSocketTimeout(Math.toIntExact(socketTimeout.getMillis()))); diff --git a/x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/slm/SnapshotLifecycleServiceTests.java b/x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/slm/SnapshotLifecycleServiceTests.java index 10bbb64cb946..dbc34d0e0f88 100644 --- a/x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/slm/SnapshotLifecycleServiceTests.java +++ b/x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/slm/SnapshotLifecycleServiceTests.java @@ -32,6 +32,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; @@ -83,7 +84,7 @@ public class SnapshotLifecycleServiceTests extends ESTestCase { assertThat(e.getMessage(), containsString("no such repository [repo]")); } - public void testNothingScheduledWhenNotRunning() { + public void testNothingScheduledWhenNotRunning() throws InterruptedException { ClockMock clock = new ClockMock(); SnapshotLifecyclePolicyMetadata initialPolicy = SnapshotLifecyclePolicyMetadata.builder() .setPolicy(createPolicy("initial", "*/1 * * * * ?")) @@ -94,13 +95,13 @@ public class SnapshotLifecycleServiceTests extends ESTestCase { ClusterState initialState = createState(new SnapshotLifecycleMetadata( Collections.singletonMap(initialPolicy.getPolicy().getId(), initialPolicy), OperationMode.RUNNING, new SnapshotLifecycleStats())); - try (ThreadPool threadPool = new TestThreadPool("test"); - ClusterService clusterService = ClusterServiceUtils.createClusterService(initialState, threadPool); + ThreadPool threadPool = new TestThreadPool("test"); + try (ClusterService clusterService = ClusterServiceUtils.createClusterService(initialState, threadPool); SnapshotLifecycleService sls = new SnapshotLifecycleService(Settings.EMPTY, () -> new FakeSnapshotTask(e -> logger.info("triggered")), clusterService, clock)) { - + sls.offMaster(); - + SnapshotLifecyclePolicyMetadata newPolicy = SnapshotLifecyclePolicyMetadata.builder() .setPolicy(createPolicy("foo", "*/1 * * * * ?")) .setHeaders(Collections.emptyMap()) @@ -113,35 +114,36 @@ public class SnapshotLifecycleServiceTests extends ESTestCase { createState(new SnapshotLifecycleMetadata(Collections.emptyMap(), OperationMode.RUNNING, new SnapshotLifecycleStats())); ClusterState state = createState(new SnapshotLifecycleMetadata(policies, OperationMode.RUNNING, new SnapshotLifecycleStats())); - + sls.clusterChanged(new ClusterChangedEvent("1", state, emptyState)); - + // Since the service does not think it is master, it should not be triggered or scheduled assertThat(sls.getScheduler().scheduledJobIds(), equalTo(Collections.emptySet())); - + sls.onMaster(); assertThat(sls.getScheduler().scheduledJobIds(), equalTo(Collections.singleton("initial-1"))); - + state = createState(new SnapshotLifecycleMetadata(policies, OperationMode.STOPPING, new SnapshotLifecycleStats())); sls.clusterChanged(new ClusterChangedEvent("2", state, emptyState)); - + // Since the service is stopping, jobs should have been cancelled assertThat(sls.getScheduler().scheduledJobIds(), equalTo(Collections.emptySet())); - + state = createState(new SnapshotLifecycleMetadata(policies, OperationMode.STOPPED, new SnapshotLifecycleStats())); sls.clusterChanged(new ClusterChangedEvent("3", state, emptyState)); - + // Since the service is stopped, jobs should have been cancelled assertThat(sls.getScheduler().scheduledJobIds(), equalTo(Collections.emptySet())); - + // No jobs should be scheduled when service is closed state = createState(new SnapshotLifecycleMetadata(policies, OperationMode.RUNNING, new SnapshotLifecycleStats())); sls.close(); sls.onMaster(); sls.clusterChanged(new ClusterChangedEvent("1", state, emptyState)); assertThat(sls.getScheduler().scheduledJobIds(), equalTo(Collections.emptySet())); - + } finally { threadPool.shutdownNow(); + threadPool.awaitTermination(10, TimeUnit.SECONDS); } } @@ -154,8 +156,8 @@ public class SnapshotLifecycleServiceTests extends ESTestCase { ClockMock clock = new ClockMock(); final AtomicInteger triggerCount = new AtomicInteger(0); final AtomicReference> trigger = new AtomicReference<>(e -> triggerCount.incrementAndGet()); - try (ThreadPool threadPool = new TestThreadPool("test"); - ClusterService clusterService = ClusterServiceUtils.createClusterService(threadPool); + ThreadPool threadPool = new TestThreadPool("test"); + try (ClusterService clusterService = ClusterServiceUtils.createClusterService(threadPool); SnapshotLifecycleService sls = new SnapshotLifecycleService(Settings.EMPTY, () -> new FakeSnapshotTask(e -> trigger.get().accept(e)), clusterService, clock)) { @@ -250,8 +252,9 @@ public class SnapshotLifecycleServiceTests extends ESTestCase { // Signify becoming non-master, the jobs should all be cancelled sls.offMaster(); assertThat(sls.getScheduler().scheduledJobIds(), equalTo(Collections.emptySet())); - + } finally { threadPool.shutdownNow(); + threadPool.awaitTermination(10, TimeUnit.SECONDS); } } @@ -262,8 +265,8 @@ public class SnapshotLifecycleServiceTests extends ESTestCase { ClockMock clock = new ClockMock(); final AtomicInteger triggerCount = new AtomicInteger(0); final AtomicReference> trigger = new AtomicReference<>(e -> triggerCount.incrementAndGet()); - try (ThreadPool threadPool = new TestThreadPool("test"); - ClusterService clusterService = ClusterServiceUtils.createClusterService(threadPool); + ThreadPool threadPool = new TestThreadPool("test"); + try (ClusterService clusterService = ClusterServiceUtils.createClusterService(threadPool); SnapshotLifecycleService sls = new SnapshotLifecycleService(Settings.EMPTY, () -> new FakeSnapshotTask(e -> trigger.get().accept(e)), clusterService, clock)) { sls.onMaster(); @@ -304,8 +307,9 @@ public class SnapshotLifecycleServiceTests extends ESTestCase { sls.offMaster(); assertThat(sls.getScheduler().scheduledJobIds(), equalTo(Collections.emptySet())); - + } finally { threadPool.shutdownNow(); + threadPool.awaitTermination(10, TimeUnit.SECONDS); } } diff --git a/x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/slm/SnapshotRetentionServiceTests.java b/x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/slm/SnapshotRetentionServiceTests.java index 28cf4f2f71e8..c2503077f0b0 100644 --- a/x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/slm/SnapshotRetentionServiceTests.java +++ b/x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/slm/SnapshotRetentionServiceTests.java @@ -26,6 +26,7 @@ import org.elasticsearch.xpack.core.watcher.watch.ClockMock; import java.util.Collections; import java.util.HashSet; import java.util.Set; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; @@ -42,13 +43,13 @@ public class SnapshotRetentionServiceTests extends ESTestCase { clusterSettings = new ClusterSettings(Settings.EMPTY, internalSettings); } - public void testJobsAreScheduled() { + public void testJobsAreScheduled() throws InterruptedException { final DiscoveryNode discoveryNode = new DiscoveryNode("node", ESTestCase.buildNewFakeTransportAddress(), Collections.emptyMap(), DiscoveryNodeRole.BUILT_IN_ROLES, Version.CURRENT); ClockMock clock = new ClockMock(); - try (ThreadPool threadPool = new TestThreadPool("test"); - ClusterService clusterService = ClusterServiceUtils.createClusterService(threadPool, discoveryNode, clusterSettings); + ThreadPool threadPool = new TestThreadPool("test"); + try (ClusterService clusterService = ClusterServiceUtils.createClusterService(threadPool, discoveryNode, clusterSettings); SnapshotRetentionService service = new SnapshotRetentionService(Settings.EMPTY, FakeRetentionTask::new, clusterService, clock)) { assertThat(service.getScheduler().jobCount(), equalTo(0)); @@ -75,33 +76,34 @@ public class SnapshotRetentionServiceTests extends ESTestCase { } } - public void testManualTriggering() { + public void testManualTriggering() throws InterruptedException { final DiscoveryNode discoveryNode = new DiscoveryNode("node", ESTestCase.buildNewFakeTransportAddress(), Collections.emptyMap(), DiscoveryNodeRole.BUILT_IN_ROLES, Version.CURRENT); ClockMock clock = new ClockMock(); AtomicInteger invoked = new AtomicInteger(0); - - try (ThreadPool threadPool = new TestThreadPool("test"); - ClusterService clusterService = ClusterServiceUtils.createClusterService(threadPool, discoveryNode, clusterSettings); + + ThreadPool threadPool = new TestThreadPool("test"); + try (ClusterService clusterService = ClusterServiceUtils.createClusterService(threadPool, discoveryNode, clusterSettings); SnapshotRetentionService service = new SnapshotRetentionService(Settings.EMPTY, () -> new FakeRetentionTask(event -> { assertThat(event.getJobName(), equalTo(SnapshotRetentionService.SLM_RETENTION_MANUAL_JOB_ID)); invoked.incrementAndGet(); }), clusterService, clock)) { - + service.onMaster(); service.triggerRetention(); assertThat(invoked.get(), equalTo(1)); - + service.offMaster(); service.triggerRetention(); assertThat(invoked.get(), equalTo(1)); - + service.onMaster(); service.triggerRetention(); assertThat(invoked.get(), equalTo(2)); - + } finally { threadPool.shutdownNow(); + threadPool.awaitTermination(10, TimeUnit.SECONDS); } } diff --git a/x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/slm/SnapshotRetentionTaskTests.java b/x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/slm/SnapshotRetentionTaskTests.java index 8dea5e2b6adb..ede01f35b04b 100644 --- a/x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/slm/SnapshotRetentionTaskTests.java +++ b/x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/slm/SnapshotRetentionTaskTests.java @@ -154,8 +154,8 @@ public class SnapshotRetentionTaskTests extends ESTestCase { } private void retentionTaskTest(final boolean deletionSuccess) throws Exception { - try (ThreadPool threadPool = new TestThreadPool("slm-test"); - ClusterService clusterService = ClusterServiceUtils.createClusterService(threadPool); + ThreadPool threadPool = new TestThreadPool("slm-test"); + try (ClusterService clusterService = ClusterServiceUtils.createClusterService(threadPool); Client noOpClient = new NoOpClient("slm-test")) { final String policyId = "policy"; @@ -222,7 +222,7 @@ public class SnapshotRetentionTaskTests extends ESTestCase { boolean historySuccess = historyLatch.await(10, TimeUnit.SECONDS); assertThat("expected history entries for 1 snapshot deletions", historySuccess, equalTo(true)); assertThat(deletedSnapshotsInHistory, contains(eligibleSnapshot.snapshotId().getName())); - + } finally { threadPool.shutdownNow(); threadPool.awaitTermination(10, TimeUnit.SECONDS); } @@ -237,8 +237,8 @@ public class SnapshotRetentionTaskTests extends ESTestCase { } private void timeBoundedDeletion(final boolean deletionSuccess) throws Exception { - try (ThreadPool threadPool = new TestThreadPool("slm-test"); - ClusterService clusterService = ClusterServiceUtils.createClusterService(threadPool); + ThreadPool threadPool = new TestThreadPool("slm-test"); + try (ClusterService clusterService = ClusterServiceUtils.createClusterService(threadPool); Client noOpClient = new NoOpClient("slm-test")) { final String policyId = "policy"; @@ -321,7 +321,7 @@ public class SnapshotRetentionTaskTests extends ESTestCase { boolean historySuccess = historyLatch.await(10, TimeUnit.SECONDS); assertThat("expected history entries for 2 snapshot deletions", historySuccess, equalTo(true)); assertThat(deletedSnapshotsInHistory, containsInAnyOrder(snap1.snapshotId().getName(), snap2.snapshotId().getName())); - + } finally { threadPool.shutdownNow(); threadPool.awaitTermination(10, TimeUnit.SECONDS); } @@ -374,8 +374,8 @@ public class SnapshotRetentionTaskTests extends ESTestCase { } private void doTestSkipDuringMode(OperationMode mode) throws Exception { - try (ThreadPool threadPool = new TestThreadPool("slm-test"); - ClusterService clusterService = ClusterServiceUtils.createClusterService(threadPool); + ThreadPool threadPool = new TestThreadPool("slm-test"); + try (ClusterService clusterService = ClusterServiceUtils.createClusterService(threadPool); Client noOpClient = new NoOpClient("slm-test")) { final String policyId = "policy"; final String repoId = "repo"; @@ -398,7 +398,7 @@ public class SnapshotRetentionTaskTests extends ESTestCase { long time = System.currentTimeMillis(); task.triggered(new SchedulerEngine.Event(SnapshotRetentionService.SLM_RETENTION_JOB_ID, time, time)); - + } finally { threadPool.shutdownNow(); threadPool.awaitTermination(10, TimeUnit.SECONDS); } @@ -413,33 +413,35 @@ public class SnapshotRetentionTaskTests extends ESTestCase { } private void doTestRunManuallyDuringMode(OperationMode mode) throws Exception { - try (ThreadPool threadPool = new TestThreadPool("slm-test"); - ClusterService clusterService = ClusterServiceUtils.createClusterService(threadPool); + ThreadPool threadPool = new TestThreadPool("slm-test"); + try (ClusterService clusterService = ClusterServiceUtils.createClusterService(threadPool); Client noOpClient = new NoOpClient("slm-test")) { final String policyId = "policy"; final String repoId = "repo"; SnapshotLifecyclePolicy policy = new SnapshotLifecyclePolicy(policyId, "snap", "1 * * * * ?", repoId, null, new SnapshotRetentionConfiguration(TimeValue.timeValueDays(30), null, null)); - + ClusterState state = createState(mode, policy); ClusterServiceUtils.setState(clusterService, state); - + AtomicBoolean retentionWasRun = new AtomicBoolean(false); MockSnapshotRetentionTask task = new MockSnapshotRetentionTask(noOpClient, clusterService, - new SnapshotLifecycleTaskTests.VerifyingHistoryStore(noOpClient, ZoneOffset.UTC, (historyItem) -> { }), + new SnapshotLifecycleTaskTests.VerifyingHistoryStore(noOpClient, ZoneOffset.UTC, (historyItem) -> { + }), threadPool, () -> { retentionWasRun.set(true); return Collections.emptyMap(); }, - (deletionPolicyId, repo, snapId, slmStats, listener) -> { }, + (deletionPolicyId, repo, snapId, slmStats, listener) -> { + }, System::nanoTime); - + long time = System.currentTimeMillis(); task.triggered(new SchedulerEngine.Event(SnapshotRetentionService.SLM_RETENTION_MANUAL_JOB_ID, time, time)); - + assertTrue("retention should be run manually even if SLM is disabled", retentionWasRun.get()); - + } finally { threadPool.shutdownNow(); threadPool.awaitTermination(10, TimeUnit.SECONDS); } diff --git a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/SecurityTests.java b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/SecurityTests.java index 99d101744e65..2d663ea619f9 100644 --- a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/SecurityTests.java +++ b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/SecurityTests.java @@ -49,9 +49,8 @@ import org.elasticsearch.xpack.security.audit.AuditTrailService; import org.elasticsearch.xpack.security.audit.logfile.LoggingAuditTrail; import org.elasticsearch.xpack.security.authc.Realms; import org.hamcrest.Matchers; -import org.junit.Before; +import org.junit.After; -import java.io.IOException; import java.util.Arrays; import java.util.Collection; import java.util.Collections; @@ -143,13 +142,9 @@ public class SecurityTests extends ESTestCase { return null; } - @Before - public void cleanup() throws IOException { - if (threadContext != null) { - threadContext.stashContext(); - threadContext.close(); - threadContext = null; - } + @After + public void cleanup() { + threadContext = null; } public void testCustomRealmExtension() throws Exception { diff --git a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptorTests.java b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptorTests.java index 89f33f809bef..11d9e6c1cd4e 100644 --- a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptorTests.java +++ b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptorTests.java @@ -399,51 +399,50 @@ public class SecurityServerTransportInterceptorTests extends ESTestCase { } public void testContextRestoreResponseHandlerRestoreOriginalContext() throws Exception { - try (ThreadContext threadContext = new ThreadContext(Settings.EMPTY)) { - threadContext.putTransient("foo", "bar"); - threadContext.putHeader("key", "value"); - TransportResponseHandler handler; - try (ThreadContext.StoredContext ignore = threadContext.stashContext()) { - threadContext.putTransient("foo", "different_bar"); - threadContext.putHeader("key", "value2"); - handler = new TransportService.ContextRestoreResponseHandler<>(threadContext.newRestorableContext(true), - new TransportResponseHandler() { + ThreadContext threadContext = new ThreadContext(Settings.EMPTY); + threadContext.putTransient("foo", "bar"); + threadContext.putHeader("key", "value"); + TransportResponseHandler handler; + try (ThreadContext.StoredContext ignore = threadContext.stashContext()) { + threadContext.putTransient("foo", "different_bar"); + threadContext.putHeader("key", "value2"); + handler = new TransportService.ContextRestoreResponseHandler<>(threadContext.newRestorableContext(true), + new TransportResponseHandler() { - @Override - public Empty read(StreamInput in) { - return Empty.INSTANCE; - } + @Override + public Empty read(StreamInput in) { + return Empty.INSTANCE; + } - @Override - public void handleResponse(Empty response) { - assertEquals("different_bar", threadContext.getTransient("foo")); - assertEquals("value2", threadContext.getHeader("key")); - } + @Override + public void handleResponse(Empty response) { + assertEquals("different_bar", threadContext.getTransient("foo")); + assertEquals("value2", threadContext.getHeader("key")); + } - @Override - public void handleException(TransportException exp) { - assertEquals("different_bar", threadContext.getTransient("foo")); - assertEquals("value2", threadContext.getHeader("key")); - } + @Override + public void handleException(TransportException exp) { + assertEquals("different_bar", threadContext.getTransient("foo")); + assertEquals("value2", threadContext.getHeader("key")); + } - @Override - public String executor() { - return null; - } - }); - } - - assertEquals("bar", threadContext.getTransient("foo")); - assertEquals("value", threadContext.getHeader("key")); - handler.handleResponse(null); - - assertEquals("bar", threadContext.getTransient("foo")); - assertEquals("value", threadContext.getHeader("key")); - handler.handleException(null); - - assertEquals("bar", threadContext.getTransient("foo")); - assertEquals("value", threadContext.getHeader("key")); + @Override + public String executor() { + return null; + } + }); } + + assertEquals("bar", threadContext.getTransient("foo")); + assertEquals("value", threadContext.getHeader("key")); + handler.handleResponse(null); + + assertEquals("bar", threadContext.getTransient("foo")); + assertEquals("value", threadContext.getHeader("key")); + handler.handleException(null); + + assertEquals("bar", threadContext.getTransient("foo")); + assertEquals("value", threadContext.getHeader("key")); } private String[] randomRoles() {