Skip to content

ThreadPool and ThreadContext are not closeable (#43249) #49273

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Nov 19, 2019
Merged

ThreadPool and ThreadContext are not closeable (#43249) #49273

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
1cdf795
ThreadPool and ThreadContext are not closeable (#43249)
jaymode Nov 18, 2019
b3f9fb7
handle rejected execution exception
jaymode Nov 19, 2019
d954d39
unused import
jaymode Nov 19, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
131 changes: 64 additions & 67 deletions qa/logging-config/src/test/java/org/elasticsearch/common/logging/JsonLoggerTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -272,80 +272,77 @@ public void testJsonInStacktraceMessageIsSplitted() throws IOException {
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");

final Path path = PathUtils.get(System.getProperty("es.logs.base_path"),
System.getProperty("es.logs.cluster_name") + "_deprecated.json");
try (Stream<Map<String, String>> stream = JsonLogsStream.mapStreamFrom(path)) {
List<Map<String, String>> 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);
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<Map<String, String>> stream = JsonLogsStream.mapStreamFrom(path)) {
List<Map<String, String>> 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);
}


// 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");

final Path path = PathUtils.get(System.getProperty("es.logs.base_path"),
System.getProperty("es.logs.cluster_name") + "_deprecated.json");
try (Stream<Map<String, String>> stream = JsonLogsStream.mapStreamFrom(path)) {
List<Map<String, String>> 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);
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<Map<String, String>> stream = JsonLogsStream.mapStreamFrom(path)) {
List<Map<String, String>> 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);
}
}

Expand Down
1 change: 0 additions & 1 deletion server/src/main/java/org/elasticsearch/common/logging/DeprecationLogger.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -262,7 +262,6 @@ public Void run() {

public String getXOpaqueId(Set<ThreadContext> 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))
Expand Down
13 changes: 2 additions & 11 deletions server/src/main/java/org/elasticsearch/common/util/concurrent/EsThreadPoolExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,17 +106,8 @@ protected void afterExecute(Runnable r, Throwable t) {
}

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;
}

Expand Down
Loading