diff options
| author | Simon Willnauer <simonw@apache.org> | 2015-12-23 14:46:54 +0100 |
|---|---|---|
| committer | Simon Willnauer <simonw@apache.org> | 2016-01-13 11:53:32 +0100 |
| commit | 574d1b35b35e2eb511cd42ab32d3d945b9852e47 (patch) | |
| tree | 4883c65a88a4fad8ba4c20d7992c7f42bacbe188 /core/src/test/java/org/elasticsearch | |
| parent | a2796b555f5c174f78303a2ce0c5a6439259f84b (diff) | |
Replace ContextAndHeaders with a ThreadPool based ThreadLocal implementation
ContextAndHeaders has a massive impact on the core infrastructure since it has to
be manually passed on to all relevant places across threads/network calls etc. For the same reason
it's also very error prone and easily forgotten on potentially relevant APIs.
The new ThreadContext is associated with a ThreadPool (node or transport client) and ensures that
headers and context registered on a current thread are inherited to new threads spawned, send across
the network to be deserialized on the receiver end as well as restored on the response handling thread
once the response is received.
Diffstat (limited to 'core/src/test/java/org/elasticsearch')
27 files changed, 463 insertions, 801 deletions
diff --git a/core/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TransportTasksActionTests.java b/core/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TransportTasksActionTests.java index 55c10aa298..635af1e51c 100644 --- a/core/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TransportTasksActionTests.java +++ b/core/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TransportTasksActionTests.java @@ -155,7 +155,7 @@ public class TransportTasksActionTests extends ESTestCase { } public NodeRequest(NodesRequest request, String nodeId) { - super(request, nodeId); + super(nodeId); requestName = request.requestName; enableTaskManager = request.enableTaskManager; } diff --git a/core/src/test/java/org/elasticsearch/client/AbstractClientHeadersTestCase.java b/core/src/test/java/org/elasticsearch/client/AbstractClientHeadersTestCase.java index b814cff520..0ecfc58de7 100644 --- a/core/src/test/java/org/elasticsearch/client/AbstractClientHeadersTestCase.java +++ b/core/src/test/java/org/elasticsearch/client/AbstractClientHeadersTestCase.java @@ -23,34 +23,21 @@ import org.elasticsearch.ExceptionsHelper; import org.elasticsearch.action.ActionListener; import org.elasticsearch.action.GenericAction; import org.elasticsearch.action.admin.cluster.reroute.ClusterRerouteAction; -import org.elasticsearch.action.admin.cluster.reroute.ClusterRerouteResponse; import org.elasticsearch.action.admin.cluster.snapshots.create.CreateSnapshotAction; -import org.elasticsearch.action.admin.cluster.snapshots.create.CreateSnapshotResponse; import org.elasticsearch.action.admin.cluster.stats.ClusterStatsAction; -import org.elasticsearch.action.admin.cluster.stats.ClusterStatsResponse; import org.elasticsearch.action.admin.indices.cache.clear.ClearIndicesCacheAction; -import org.elasticsearch.action.admin.indices.cache.clear.ClearIndicesCacheResponse; import org.elasticsearch.action.admin.indices.create.CreateIndexAction; -import org.elasticsearch.action.admin.indices.create.CreateIndexResponse; import org.elasticsearch.action.admin.indices.flush.FlushAction; -import org.elasticsearch.action.admin.indices.flush.FlushResponse; import org.elasticsearch.action.admin.indices.stats.IndicesStatsAction; -import org.elasticsearch.action.admin.indices.stats.IndicesStatsResponse; import org.elasticsearch.action.delete.DeleteAction; -import org.elasticsearch.action.delete.DeleteResponse; import org.elasticsearch.action.get.GetAction; -import org.elasticsearch.action.get.GetResponse; import org.elasticsearch.action.index.IndexAction; -import org.elasticsearch.action.index.IndexResponse; import org.elasticsearch.action.indexedscripts.delete.DeleteIndexedScriptAction; -import org.elasticsearch.action.indexedscripts.delete.DeleteIndexedScriptResponse; import org.elasticsearch.action.search.SearchAction; -import org.elasticsearch.action.search.SearchResponse; -import org.elasticsearch.client.support.Headers; import org.elasticsearch.common.settings.Settings; +import org.elasticsearch.common.util.concurrent.ThreadContext; import org.elasticsearch.test.ESTestCase; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.transport.TransportMessage; import org.junit.After; import org.junit.Before; @@ -67,8 +54,8 @@ import static org.hamcrest.Matchers.notNullValue; public abstract class AbstractClientHeadersTestCase extends ESTestCase { protected static final Settings HEADER_SETTINGS = Settings.builder() - .put(Headers.PREFIX + ".key1", "val1") - .put(Headers.PREFIX + ".key2", "val 2") + .put(ThreadContext.PREFIX + ".key1", "val1") + .put(ThreadContext.PREFIX + ".key2", "val 2") .build(); private static final GenericAction[] ACTIONS = new GenericAction[] { @@ -91,8 +78,9 @@ public abstract class AbstractClientHeadersTestCase extends ESTestCase { Settings settings = Settings.builder() .put(HEADER_SETTINGS) .put("path.home", createTempDir().toString()) + .put("name", "test-" + getTestName()) .build(); - threadPool = new ThreadPool("test-" + getTestName()); + threadPool = new ThreadPool(settings); client = buildClient(settings, ACTIONS); } @@ -113,89 +101,75 @@ public abstract class AbstractClientHeadersTestCase extends ESTestCase { // validation in the settings??? - ugly and conceptually wrong) // choosing arbitrary top level actions to test - client.prepareGet("idx", "type", "id").execute().addListener(new AssertingActionListener<GetResponse>(GetAction.NAME)); - client.prepareSearch().execute().addListener(new AssertingActionListener<SearchResponse>(SearchAction.NAME)); - client.prepareDelete("idx", "type", "id").execute().addListener(new AssertingActionListener<DeleteResponse>(DeleteAction.NAME)); - client.prepareDeleteIndexedScript("lang", "id").execute().addListener(new AssertingActionListener<DeleteIndexedScriptResponse>(DeleteIndexedScriptAction.NAME)); - client.prepareIndex("idx", "type", "id").setSource("source").execute().addListener(new AssertingActionListener<IndexResponse>(IndexAction.NAME)); + client.prepareGet("idx", "type", "id").execute().addListener(new AssertingActionListener<>(GetAction.NAME, client.threadPool())); + client.prepareSearch().execute().addListener(new AssertingActionListener<>(SearchAction.NAME, client.threadPool())); + client.prepareDelete("idx", "type", "id").execute().addListener(new AssertingActionListener<>(DeleteAction.NAME, client.threadPool())); + client.prepareDeleteIndexedScript("lang", "id").execute().addListener(new AssertingActionListener<>(DeleteIndexedScriptAction.NAME, client.threadPool())); + client.prepareIndex("idx", "type", "id").setSource("source").execute().addListener(new AssertingActionListener<>(IndexAction.NAME, client.threadPool())); // choosing arbitrary cluster admin actions to test - client.admin().cluster().prepareClusterStats().execute().addListener(new AssertingActionListener<ClusterStatsResponse>(ClusterStatsAction.NAME)); - client.admin().cluster().prepareCreateSnapshot("repo", "bck").execute().addListener(new AssertingActionListener<CreateSnapshotResponse>(CreateSnapshotAction.NAME)); - client.admin().cluster().prepareReroute().execute().addListener(new AssertingActionListener<ClusterRerouteResponse>(ClusterRerouteAction.NAME)); + client.admin().cluster().prepareClusterStats().execute().addListener(new AssertingActionListener<>(ClusterStatsAction.NAME, client.threadPool())); + client.admin().cluster().prepareCreateSnapshot("repo", "bck").execute().addListener(new AssertingActionListener<>(CreateSnapshotAction.NAME, client.threadPool())); + client.admin().cluster().prepareReroute().execute().addListener(new AssertingActionListener<>(ClusterRerouteAction.NAME, client.threadPool())); // choosing arbitrary indices admin actions to test - client.admin().indices().prepareCreate("idx").execute().addListener(new AssertingActionListener<CreateIndexResponse>(CreateIndexAction.NAME)); - client.admin().indices().prepareStats().execute().addListener(new AssertingActionListener<IndicesStatsResponse>(IndicesStatsAction.NAME)); - client.admin().indices().prepareClearCache("idx1", "idx2").execute().addListener(new AssertingActionListener<ClearIndicesCacheResponse>(ClearIndicesCacheAction.NAME)); - client.admin().indices().prepareFlush().execute().addListener(new AssertingActionListener<FlushResponse>(FlushAction.NAME)); + client.admin().indices().prepareCreate("idx").execute().addListener(new AssertingActionListener<>(CreateIndexAction.NAME, client.threadPool())); + client.admin().indices().prepareStats().execute().addListener(new AssertingActionListener<>(IndicesStatsAction.NAME, client.threadPool())); + client.admin().indices().prepareClearCache("idx1", "idx2").execute().addListener(new AssertingActionListener<>(ClearIndicesCacheAction.NAME, client.threadPool())); + client.admin().indices().prepareFlush().execute().addListener(new AssertingActionListener<>(FlushAction.NAME, client.threadPool())); } public void testOverideHeader() throws Exception { String key1Val = randomAsciiOfLength(5); - Map<String, Object> expected = new HashMap<>(); + Map<String, String> expected = new HashMap<>(); expected.put("key1", key1Val); expected.put("key2", "val 2"); - + client.threadPool().getThreadContext().putHeader("key1", key1Val); client.prepareGet("idx", "type", "id") - .putHeader("key1", key1Val) - .execute().addListener(new AssertingActionListener<GetResponse>(GetAction.NAME, expected)); + .execute().addListener(new AssertingActionListener<>(GetAction.NAME, expected, client.threadPool())); client.admin().cluster().prepareClusterStats() - .putHeader("key1", key1Val) - .execute().addListener(new AssertingActionListener<ClusterStatsResponse>(ClusterStatsAction.NAME, expected)); + .execute().addListener(new AssertingActionListener<>(ClusterStatsAction.NAME, expected, client.threadPool())); client.admin().indices().prepareCreate("idx") - .putHeader("key1", key1Val) - .execute().addListener(new AssertingActionListener<CreateIndexResponse>(CreateIndexAction.NAME, expected)); + .execute().addListener(new AssertingActionListener<>(CreateIndexAction.NAME, expected, client.threadPool())); } - protected static void assertHeaders(Map<String, Object> headers, Map<String, Object> expected) { - assertThat(headers, notNullValue()); - assertThat(headers.size(), is(expected.size())); - for (Map.Entry<String, Object> expectedEntry : expected.entrySet()) { - assertThat(headers.get(expectedEntry.getKey()), equalTo(expectedEntry.getValue())); + protected static void assertHeaders(Map<String, String> headers, Map<String, String> expected) { + assertNotNull(headers); + assertEquals(expected.size(), headers.size()); + for (Map.Entry<String, String> expectedEntry : expected.entrySet()) { + assertEquals(headers.get(expectedEntry.getKey()), expectedEntry.getValue()); } } - protected static void assertHeaders(TransportMessage<?> message) { - assertHeaders(message, HEADER_SETTINGS.getAsSettings(Headers.PREFIX).getAsStructuredMap()); - } - - protected static void assertHeaders(TransportMessage<?> message, Map<String, Object> expected) { - assertThat(message.getHeaders(), notNullValue()); - assertThat(message.getHeaders().size(), is(expected.size())); - for (Map.Entry<String, Object> expectedEntry : expected.entrySet()) { - assertThat(message.getHeader(expectedEntry.getKey()), equalTo(expectedEntry.getValue())); - } + protected static void assertHeaders(ThreadPool pool) { + assertHeaders(pool.getThreadContext().getHeaders(), (Map)HEADER_SETTINGS.getAsSettings(ThreadContext.PREFIX).getAsStructuredMap()); } public static class InternalException extends Exception { private final String action; - private final Map<String, Object> headers; - public InternalException(String action, TransportMessage<?> message) { + public InternalException(String action) { this.action = action; - this.headers = new HashMap<>(); - for (String key : message.getHeaders()) { - headers.put(key, message.getHeader(key)); - } } } protected static class AssertingActionListener<T> implements ActionListener<T> { private final String action; - private final Map<String, Object> expectedHeaders; + private final Map<String, String> expectedHeaders; + private final ThreadPool pool; - public AssertingActionListener(String action) { - this(action, HEADER_SETTINGS.getAsSettings(Headers.PREFIX).getAsStructuredMap()); + public AssertingActionListener(String action, ThreadPool pool) { + this(action, (Map)HEADER_SETTINGS.getAsSettings(ThreadContext.PREFIX).getAsStructuredMap(), pool); } - public AssertingActionListener(String action, Map<String, Object> expectedHeaders) { + public AssertingActionListener(String action, Map<String, String> expectedHeaders, ThreadPool pool) { this.action = action; this.expectedHeaders = expectedHeaders; + this.pool = pool; } @Override @@ -208,7 +182,7 @@ public abstract class AbstractClientHeadersTestCase extends ESTestCase { Throwable e = unwrap(t, InternalException.class); assertThat("expected action [" + action + "] to throw an internal exception", e, notNullValue()); assertThat(action, equalTo(((InternalException) e).action)); - Map<String, Object> headers = ((InternalException) e).headers; + Map<String, String> headers = pool.getThreadContext().getHeaders(); assertHeaders(headers, expectedHeaders); } diff --git a/core/src/test/java/org/elasticsearch/client/node/NodeClientHeadersTests.java b/core/src/test/java/org/elasticsearch/client/node/NodeClientHeadersTests.java index e7ba8de0f9..f69c8f2da0 100644 --- a/core/src/test/java/org/elasticsearch/client/node/NodeClientHeadersTests.java +++ b/core/src/test/java/org/elasticsearch/client/node/NodeClientHeadersTests.java @@ -27,7 +27,6 @@ import org.elasticsearch.action.support.ActionFilters; import org.elasticsearch.action.support.TransportAction; import org.elasticsearch.client.AbstractClientHeadersTestCase; import org.elasticsearch.client.Client; -import org.elasticsearch.client.support.Headers; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.tasks.Task; import org.elasticsearch.tasks.TaskManager; @@ -46,9 +45,8 @@ public class NodeClientHeadersTests extends AbstractClientHeadersTestCase { @Override protected Client buildClient(Settings headersSettings, GenericAction[] testedActions) { Settings settings = HEADER_SETTINGS; - Headers headers = new Headers(settings); Actions actions = new Actions(settings, threadPool, testedActions); - return new NodeClient(settings, threadPool, headers, actions); + return new NodeClient(settings, threadPool, actions); } private static class Actions extends HashMap<GenericAction, TransportAction> { @@ -68,7 +66,7 @@ public class NodeClientHeadersTests extends AbstractClientHeadersTestCase { @Override protected void doExecute(ActionRequest request, ActionListener listener) { - listener.onFailure(new InternalException(actionName, request)); + listener.onFailure(new InternalException(actionName)); } } diff --git a/core/src/test/java/org/elasticsearch/client/transport/TransportClientHeadersTests.java b/core/src/test/java/org/elasticsearch/client/transport/TransportClientHeadersTests.java index f127ae2837..c364e640a2 100644 --- a/core/src/test/java/org/elasticsearch/client/transport/TransportClientHeadersTests.java +++ b/core/src/test/java/org/elasticsearch/client/transport/TransportClientHeadersTests.java @@ -134,30 +134,30 @@ public class TransportClientHeadersTests extends AbstractClientHeadersTestCase { @Override @SuppressWarnings("unchecked") public <T extends TransportResponse> void sendRequest(DiscoveryNode node, String action, TransportRequest request, TransportRequestOptions options, TransportResponseHandler<T> handler) { if (TransportLivenessAction.NAME.equals(action)) { - assertHeaders(request); + assertHeaders(threadPool); ((TransportResponseHandler<LivenessResponse>) handler).handleResponse(new LivenessResponse(ClusterName.DEFAULT, node)); return; } if (ClusterStateAction.NAME.equals(action)) { - assertHeaders(request); + assertHeaders(threadPool); ClusterName cluster1 = new ClusterName("cluster1"); ((TransportResponseHandler<ClusterStateResponse>) handler).handleResponse(new ClusterStateResponse(cluster1, state(cluster1))); clusterStateLatch.countDown(); return; } - handler.handleException(new TransportException("", new InternalException(action, request))); + handler.handleException(new TransportException("", new InternalException(action))); } @Override public boolean nodeConnected(DiscoveryNode node) { - assertThat((LocalTransportAddress) node.getAddress(), equalTo(address)); + assertThat(node.getAddress(), equalTo(address)); return true; } @Override public void connectToNode(DiscoveryNode node) throws ConnectTransportException { - assertThat((LocalTransportAddress) node.getAddress(), equalTo(address)); + assertThat(node.getAddress(), equalTo(address)); } } diff --git a/core/src/test/java/org/elasticsearch/client/transport/TransportClientNodesServiceTests.java b/core/src/test/java/org/elasticsearch/client/transport/TransportClientNodesServiceTests.java index 72ace64d9e..e6ea0410a5 100644 --- a/core/src/test/java/org/elasticsearch/client/transport/TransportClientNodesServiceTests.java +++ b/core/src/test/java/org/elasticsearch/client/transport/TransportClientNodesServiceTests.java @@ -21,7 +21,6 @@ package org.elasticsearch.client.transport; import org.elasticsearch.Version; import org.elasticsearch.action.ActionListener; -import org.elasticsearch.client.support.Headers; import org.elasticsearch.cluster.ClusterName; import org.elasticsearch.cluster.node.DiscoveryNode; import org.elasticsearch.common.settings.Settings; @@ -74,7 +73,7 @@ public class TransportClientNodesServiceTests extends ESTestCase { }; transportService = new TransportService(Settings.EMPTY, transport, threadPool); transportService.start(); - transportClientNodesService = new TransportClientNodesService(Settings.EMPTY, ClusterName.DEFAULT, transportService, threadPool, Headers.EMPTY, Version.CURRENT); + transportClientNodesService = new TransportClientNodesService(Settings.EMPTY, ClusterName.DEFAULT, transportService, threadPool, Version.CURRENT); nodesCount = randomIntBetween(1, 10); for (int i = 0; i < nodesCount; i++) { diff --git a/core/src/test/java/org/elasticsearch/common/network/NetworkModuleTests.java b/core/src/test/java/org/elasticsearch/common/network/NetworkModuleTests.java index 798e82a979..c181d89356 100644 --- a/core/src/test/java/org/elasticsearch/common/network/NetworkModuleTests.java +++ b/core/src/test/java/org/elasticsearch/common/network/NetworkModuleTests.java @@ -82,7 +82,7 @@ public class NetworkModuleTests extends ModuleTestCase { static class FakeRestHandler extends BaseRestHandler { public FakeRestHandler() { - super(null, null, null); + super(null, null); } @Override protected void handleRequest(RestRequest request, RestChannel channel, Client client) throws Exception {} diff --git a/core/src/test/java/org/elasticsearch/common/util/concurrent/EsExecutorsTests.java b/core/src/test/java/org/elasticsearch/common/util/concurrent/EsExecutorsTests.java index b59c8dd1cb..fabfc6a94e 100644 --- a/core/src/test/java/org/elasticsearch/common/util/concurrent/EsExecutorsTests.java +++ b/core/src/test/java/org/elasticsearch/common/util/concurrent/EsExecutorsTests.java @@ -20,6 +20,7 @@ package org.elasticsearch.common.util.concurrent; import org.elasticsearch.ExceptionsHelper; +import org.elasticsearch.common.settings.Settings; import org.elasticsearch.test.ESTestCase; import org.hamcrest.Matcher; @@ -38,12 +39,13 @@ import static org.hamcrest.Matchers.lessThan; */ public class EsExecutorsTests extends ESTestCase { + private final ThreadContext threadContext = new ThreadContext(Settings.EMPTY); private TimeUnit randomTimeUnit() { return TimeUnit.values()[between(0, TimeUnit.values().length - 1)]; } public void testFixedForcedExecution() throws Exception { - EsThreadPoolExecutor executor = EsExecutors.newFixed(getTestName(), 1, 1, EsExecutors.daemonThreadFactory("test")); + EsThreadPoolExecutor executor = EsExecutors.newFixed(getTestName(), 1, 1, EsExecutors.daemonThreadFactory("test"), threadContext); final CountDownLatch wait = new CountDownLatch(1); final CountDownLatch exec1Wait = new CountDownLatch(1); @@ -105,7 +107,7 @@ public class EsExecutorsTests extends ESTestCase { } public void testFixedRejected() throws Exception { - EsThreadPoolExecutor executor = EsExecutors.newFixed(getTestName(), 1, 1, EsExecutors.daemonThreadFactory("test")); + EsThreadPoolExecutor executor = EsExecutors.newFixed(getTestName(), 1, 1, EsExecutors.daemonThreadFactory("test"), threadContext); final CountDownLatch wait = new CountDownLatch(1); final CountDownLatch exec1Wait = new CountDownLatch(1); @@ -163,7 +165,7 @@ public class EsExecutorsTests extends ESTestCase { final int max = between(min + 1, 6); final ThreadBarrier barrier = new ThreadBarrier(max + 1); - ThreadPoolExecutor pool = EsExecutors.newScaling(getTestName(), min, max, between(1, 100), randomTimeUnit(), EsExecutors.daemonThreadFactory("test")); + ThreadPoolExecutor pool = EsExecutors.newScaling(getTestName(), min, max, between(1, 100), randomTimeUnit(), EsExecutors.daemonThreadFactory("test"), threadContext); assertThat("Min property", pool.getCorePoolSize(), equalTo(min)); assertThat("Max property", pool.getMaximumPoolSize(), equalTo(max)); @@ -199,7 +201,7 @@ public class EsExecutorsTests extends ESTestCase { final int max = between(min + 1, 6); final ThreadBarrier barrier = new ThreadBarrier(max + 1); - final ThreadPoolExecutor pool = EsExecutors.newScaling(getTestName(), min, max, between(1, 100), TimeUnit.MILLISECONDS, EsExecutors.daemonThreadFactory("test")); + final ThreadPoolExecutor pool = EsExecutors.newScaling(getTestName(), min, max, between(1, 100), TimeUnit.MILLISECONDS, EsExecutors.daemonThreadFactory("test"), threadContext); assertThat("Min property", pool.getCorePoolSize(), equalTo(min)); assertThat("Max property", pool.getMaximumPoolSize(), equalTo(max)); @@ -242,7 +244,7 @@ public class EsExecutorsTests extends ESTestCase { int queue = between(0, 100); int actions = queue + pool; final CountDownLatch latch = new CountDownLatch(1); - EsThreadPoolExecutor executor = EsExecutors.newFixed(getTestName(), pool, queue, EsExecutors.daemonThreadFactory("dummy")); + EsThreadPoolExecutor executor = EsExecutors.newFixed(getTestName(), pool, queue, EsExecutors.daemonThreadFactory("dummy"), threadContext); try { for (int i = 0; i < actions; i++) { executor.execute(new Runnable() { @@ -321,4 +323,42 @@ public class EsExecutorsTests extends ESTestCase { assertThat(message, containsString("completed tasks = " + actions)); } } + + public void testInheritContext() throws InterruptedException { + int pool = between(1, 10); + int queue = between(0, 100); + final CountDownLatch latch = new CountDownLatch(1); + final CountDownLatch executed = new CountDownLatch(1); + + threadContext.putHeader("foo", "bar"); + final Integer one = new Integer(1); + threadContext.putTransient("foo", one); + EsThreadPoolExecutor executor = EsExecutors.newFixed(getTestName(), pool, queue, EsExecutors.daemonThreadFactory("dummy"), threadContext); + try { + executor.execute(new Runnable() { + @Override + public void run() { + try { + latch.await(); + } catch (InterruptedException e) { + fail(); + } + assertEquals(threadContext.getHeader("foo"), "bar"); + assertSame(threadContext.getTransient("foo"), one); + assertNull(threadContext.getHeader("bar")); + assertNull(threadContext.getTransient("bar")); + executed.countDown(); + } + }); + threadContext.putTransient("bar", "boom"); + threadContext.putHeader("bar", "boom"); + latch.countDown(); + executed.await(); + + } finally { + latch.countDown(); + terminate(executor); + } + + } } diff --git a/core/src/test/java/org/elasticsearch/common/util/concurrent/PrioritizedExecutorsTests.java b/core/src/test/java/org/elasticsearch/common/util/concurrent/PrioritizedExecutorsTests.java index 685e06afb1..50b7d5f775 100644 --- a/core/src/test/java/org/elasticsearch/common/util/concurrent/PrioritizedExecutorsTests.java +++ b/core/src/test/java/org/elasticsearch/common/util/concurrent/PrioritizedExecutorsTests.java @@ -19,6 +19,7 @@ package org.elasticsearch.common.util.concurrent; import org.elasticsearch.common.Priority; +import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.test.ESTestCase; import org.elasticsearch.threadpool.ThreadPool; @@ -43,6 +44,9 @@ import static org.hamcrest.Matchers.is; * */ public class PrioritizedExecutorsTests extends ESTestCase { + + private final ThreadContext holder = new ThreadContext(Settings.EMPTY); + public void testPriorityQueue() throws Exception { PriorityBlockingQueue<Priority> queue = new PriorityBlockingQueue<>(); List<Priority> priorities = Arrays.asList(Priority.values()); @@ -63,7 +67,7 @@ public class PrioritizedExecutorsTests extends ESTestCase { } public void testSubmitPrioritizedExecutorWithRunnables() throws Exception { - ExecutorService executor = EsExecutors.newSinglePrioritizing(getTestName(), EsExecutors.daemonThreadFactory(getTestName())); + ExecutorService executor = EsExecutors.newSinglePrioritizing(getTestName(), EsExecutors.daemonThreadFactory(getTestName()), holder); List<Integer> results = new ArrayList<>(8); CountDownLatch awaitingLatch = new CountDownLatch(1); CountDownLatch finishedLatch = new CountDownLatch(8); @@ -92,7 +96,7 @@ public class PrioritizedExecutorsTests extends ESTestCase { } public void testExecutePrioritizedExecutorWithRunnables() throws Exception { - ExecutorService executor = EsExecutors.newSinglePrioritizing(getTestName(), EsExecutors.daemonThreadFactory(getTestName())); + ExecutorService executor = EsExecutors.newSinglePrioritizing(getTestName(), EsExecutors.daemonThreadFactory(getTestName()), holder); List<Integer> results = new ArrayList<>(8); CountDownLatch awaitingLatch = new CountDownLatch(1); CountDownLatch finishedLatch = new CountDownLatch(8); @@ -121,7 +125,7 @@ public class PrioritizedExecutorsTests extends ESTestCase { } public void testSubmitPrioritizedExecutorWithCallables() throws Exception { - ExecutorService executor = EsExecutors.newSinglePrioritizing(getTestName(), EsExecutors.daemonThreadFactory(getTestName())); + ExecutorService executor = EsExecutors.newSinglePrioritizing(getTestName(), EsExecutors.daemonThreadFactory(getTestName()), holder); List<Integer> results = new ArrayList<>(8); CountDownLatch awaitingLatch = new CountDownLatch(1); CountDownLatch finishedLatch = new CountDownLatch(8); @@ -150,7 +154,7 @@ public class PrioritizedExecutorsTests extends ESTestCase { } public void testSubmitPrioritizedExecutorWithMixed() throws Exception { - ExecutorService executor = EsExecutors.newSinglePrioritizing(getTestName(), EsExecutors.daemonThreadFactory(getTestName())); + ExecutorService executor = EsExecutors.newSinglePrioritizing(getTestName(), EsExecutors.daemonThreadFactory(getTestName()), holder); List<Integer> results = new ArrayList<>(8); CountDownLatch awaitingLatch = new CountDownLatch(1); CountDownLatch finishedLatch = new CountDownLatch(8); @@ -180,7 +184,7 @@ public class PrioritizedExecutorsTests extends ESTestCase { public void testTimeout() throws Exception { ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor(EsExecutors.daemonThreadFactory(getTestName())); - PrioritizedEsThreadPoolExecutor executor = EsExecutors.newSinglePrioritizing(getTestName(), EsExecutors.daemonThreadFactory(getTestName())); + PrioritizedEsThreadPoolExecutor executor = EsExecutors.newSinglePrioritizing(getTestName(), EsExecutors.daemonThreadFactory(getTestName()), holder); final CountDownLatch invoked = new CountDownLatch(1); final CountDownLatch block = new CountDownLatch(1); executor.execute(new Runnable() { @@ -243,7 +247,7 @@ public class PrioritizedExecutorsTests extends ESTestCase { ThreadPool threadPool = new ThreadPool("test"); final ScheduledThreadPoolExecutor timer = (ScheduledThreadPoolExecutor) threadPool.scheduler(); final AtomicBoolean timeoutCalled = new AtomicBoolean(); - PrioritizedEsThreadPoolExecutor executor = EsExecutors.newSinglePrioritizing(getTestName(), EsExecutors.daemonThreadFactory(getTestName())); + PrioritizedEsThreadPoolExecutor executor = EsExecutors.newSinglePrioritizing(getTestName(), EsExecutors.daemonThreadFactory(getTestName()), holder); final CountDownLatch invoked = new CountDownLatch(1); executor.execute(new Runnable() { @Override diff --git a/core/src/test/java/org/elasticsearch/common/util/concurrent/ThreadContextTests.java b/core/src/test/java/org/elasticsearch/common/util/concurrent/ThreadContextTests.java new file mode 100644 index 0000000000..bbe5136f57 --- /dev/null +++ b/core/src/test/java/org/elasticsearch/common/util/concurrent/ThreadContextTests.java @@ -0,0 +1,138 @@ +/* + * Licensed to Elasticsearch under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.elasticsearch.common.util.concurrent; + +import org.elasticsearch.common.io.stream.BytesStreamOutput; +import org.elasticsearch.common.io.stream.StreamInput; +import org.elasticsearch.common.settings.Settings; +import org.elasticsearch.test.ESTestCase; + +import java.io.IOException; +import java.util.Collections; + +public class ThreadContextTests extends ESTestCase { + + public void testStashContext() { + Settings build = Settings.builder().put("request.headers.default", "1").build(); + ThreadContext threadContext = new ThreadContext(build); + threadContext.putHeader("foo", "bar"); + threadContext.putTransient("ctx.foo", new Integer(1)); + assertEquals("bar", threadContext.getHeader("foo")); + assertEquals(new Integer(1), threadContext.getTransient("ctx.foo")); + assertEquals("1", threadContext.getHeader("default")); + try (ThreadContext.StoredContext ctx = threadContext.stashContext()) { + assertNull(threadContext.getHeader("foo")); + assertNull(threadContext.getTransient("ctx.foo")); + assertEquals("1", threadContext.getHeader("default")); + } + + assertEquals("bar", threadContext.getHeader("foo")); + assertEquals(new Integer(1), threadContext.getTransient("ctx.foo")); + assertEquals("1", threadContext.getHeader("default")); + } + + public void testStoreContext() { + Settings build = Settings.builder().put("request.headers.default", "1").build(); + ThreadContext threadContext = new ThreadContext(build); + threadContext.putHeader("foo", "bar"); + threadContext.putTransient("ctx.foo", new Integer(1)); + assertEquals("bar", threadContext.getHeader("foo")); + assertEquals(new Integer(1), threadContext.getTransient("ctx.foo")); + assertEquals("1", threadContext.getHeader("default")); + ThreadContext.StoredContext storedContext = threadContext.newStoredContext(); + threadContext.putHeader("foo.bar", "baz"); + try (ThreadContext.StoredContext ctx = threadContext.stashContext()) { + assertNull(threadContext.getHeader("foo")); + assertNull(threadContext.getTransient("ctx.foo")); + assertEquals("1", threadContext.getHeader("default")); + } + + assertEquals("bar", threadContext.getHeader("foo")); + assertEquals(new Integer(1), threadContext.getTransient("ctx.foo")); + assertEquals("1", threadContext.getHeader("default")); + assertEquals("baz", threadContext.getHeader("foo.bar")); + if (randomBoolean()) { + storedContext.restore(); + } else { + storedContext.close(); + } + assertEquals("bar", threadContext.getHeader("foo")); + assertEquals(new Integer(1), threadContext.getTransient("ctx.foo")); + assertEquals("1", threadContext.getHeader("default")); + assertNull(threadContext.getHeader("foo.bar")); + } + + public void testCopyHeaders() { + Settings build = Settings.builder().put("request.headers.default", "1").build(); + ThreadContext threadContext = new ThreadContext(build); + threadContext.copyHeaders(Collections.<String,String>emptyMap().entrySet()); + threadContext.copyHeaders(Collections.<String,String>singletonMap("foo", "bar").entrySet()); + 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", new Integer(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); + threadContext.putHeader("foo", "bar"); + threadContext.putTransient("ctx.foo", new Integer(1)); + BytesStreamOutput out = new BytesStreamOutput(); + threadContext.writeTo(out); + try (ThreadContext.StoredContext ctx = threadContext.stashContext()) { + assertNull(threadContext.getHeader("foo")); + assertNull(threadContext.getTransient("ctx.foo")); + assertEquals("1", threadContext.getHeader("default")); + + threadContext.readHeaders(StreamInput.wrap(out.bytes())); + assertEquals("bar", threadContext.getHeader("foo")); + assertNull(threadContext.getTransient("ctx.foo")); + } + assertEquals("bar", threadContext.getHeader("foo")); + assertEquals(new Integer(1), threadContext.getTransient("ctx.foo")); + assertEquals("1", threadContext.getHeader("default")); + } +} diff --git a/core/src/test/java/org/elasticsearch/http/netty/NettyHttpChannelTests.java b/core/src/test/java/org/elasticsearch/http/netty/NettyHttpChannelTests.java index cb111a7198..179d1d1329 100644 --- a/core/src/test/java/org/elasticsearch/http/netty/NettyHttpChannelTests.java +++ b/core/src/test/java/org/elasticsearch/http/netty/NettyHttpChannelTests.java @@ -83,7 +83,7 @@ public class NettyHttpChannelTests extends ESTestCase { Settings settings = Settings.builder() .put(NettyHttpServerTransport.SETTING_CORS_ENABLED, true) .build(); - httpServerTransport = new NettyHttpServerTransport(settings, networkService, bigArrays); + httpServerTransport = new NettyHttpServerTransport(settings, networkService, bigArrays, threadPool); HttpRequest httpRequest = new TestHttpRequest(); httpRequest.headers().add(HttpHeaders.Names.ORIGIN, "remote"); httpRequest.headers().add(HttpHeaders.Names.USER_AGENT, "Mozilla fake"); @@ -107,7 +107,7 @@ public class NettyHttpChannelTests extends ESTestCase { .put(NettyHttpServerTransport.SETTING_CORS_ENABLED, true) .put(NettyHttpServerTransport.SETTING_CORS_ALLOW_ORIGIN, "remote-host") .build(); - httpServerTransport = new NettyHttpServerTransport(settings, networkService, bigArrays); + httpServerTransport = new NettyHttpServerTransport(settings, networkService, bigArrays, threadPool); HttpRequest httpRequest = new TestHttpRequest(); httpRequest.headers().add(HttpHeaders.Names.ORIGIN, "remote"); httpRequest.headers().add(HttpHeaders.Names.USER_AGENT, "Mozilla fake"); diff --git a/core/src/test/java/org/elasticsearch/http/netty/NettyHttpServerPipeliningTests.java b/core/src/test/java/org/elasticsearch/http/netty/NettyHttpServerPipeliningTests.java index 95cb5b46b5..6afe8a0aef 100644 --- a/core/src/test/java/org/elasticsearch/http/netty/NettyHttpServerPipeliningTests.java +++ b/core/src/test/java/org/elasticsearch/http/netty/NettyHttpServerPipeliningTests.java @@ -23,6 +23,7 @@ import org.elasticsearch.common.network.NetworkService; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.transport.InetSocketTransportAddress; import org.elasticsearch.common.util.MockBigArrays; +import org.elasticsearch.common.util.concurrent.ThreadContext; import org.elasticsearch.http.HttpServerTransport; import org.elasticsearch.http.netty.NettyHttpServerTransport.HttpChannelPipelineFactory; import org.elasticsearch.http.netty.pipelining.OrderedDownstreamChannelEvent; @@ -132,13 +133,13 @@ public class NettyHttpServerPipeliningTests extends ESTestCase { private final ExecutorService executorService; public CustomNettyHttpServerTransport(Settings settings) { - super(settings, NettyHttpServerPipeliningTests.this.networkService, NettyHttpServerPipeliningTests.this.bigArrays); + super(settings, NettyHttpServerPipeliningTests.this.networkService, NettyHttpServerPipeliningTests.this.bigArrays, NettyHttpServerPipeliningTests.this.threadPool); this.executorService = Executors.newFixedThreadPool(5); } @Override public ChannelPipelineFactory configureServerChannelPipelineFactory() { - return new CustomHttpChannelPipelineFactory(this, executorService); + return new CustomHttpChannelPipelineFactory(this, executorService, NettyHttpServerPipeliningTests.this.threadPool.getThreadContext()); } @Override @@ -152,8 +153,8 @@ public class NettyHttpServerPipeliningTests extends ESTestCase { private final ExecutorService executorService; - public CustomHttpChannelPipelineFactory(NettyHttpServerTransport transport, ExecutorService executorService) { - super(transport, randomBoolean()); + public CustomHttpChannelPipelineFactory(NettyHttpServerTransport transport, ExecutorService executorService, ThreadContext threadContext) { + super(transport, randomBoolean(), threadContext); this.executorService = executorService; } diff --git a/core/src/test/java/org/elasticsearch/indices/store/IndicesStoreTests.java b/core/src/test/java/org/elasticsearch/indices/store/IndicesStoreTests.java index ec6a3b3849..a7b08be39d 100644 --- a/core/src/test/java/org/elasticsearch/indices/store/IndicesStoreTests.java +++ b/core/src/test/java/org/elasticsearch/indices/store/IndicesStoreTests.java @@ -30,9 +30,12 @@ import org.elasticsearch.cluster.routing.IndexShardRoutingTable; import org.elasticsearch.cluster.routing.ShardRoutingState; import org.elasticsearch.cluster.routing.TestShardRouting; import org.elasticsearch.cluster.routing.UnassignedInfo; +import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.transport.LocalTransportAddress; import org.elasticsearch.index.shard.ShardId; import org.elasticsearch.test.ESTestCase; +import org.elasticsearch.test.cluster.TestClusterService; +import org.elasticsearch.transport.TransportService; import org.junit.Before; import java.util.Arrays; @@ -60,7 +63,7 @@ public class IndicesStoreTests extends ESTestCase { @Before public void before() { localNode = new DiscoveryNode("abc", new LocalTransportAddress("abc"), Version.CURRENT); - indicesStore = new IndicesStore(); + indicesStore = new IndicesStore(Settings.EMPTY, null, new TestClusterService(), new TransportService(null, null), null); } public void testShardCanBeDeletedNoShardRouting() throws Exception { diff --git a/core/src/test/java/org/elasticsearch/indices/template/IndexTemplateFilteringIT.java b/core/src/test/java/org/elasticsearch/indices/template/IndexTemplateFilteringIT.java index ee0f874808..8e0d5a882c 100644 --- a/core/src/test/java/org/elasticsearch/indices/template/IndexTemplateFilteringIT.java +++ b/core/src/test/java/org/elasticsearch/indices/template/IndexTemplateFilteringIT.java @@ -57,7 +57,7 @@ public class IndexTemplateFilteringIT extends ESIntegTestCase { .setTemplate("no_match") .addMapping("type3", "field3", "type=string").get(); - assertAcked(prepareCreate("test").putHeader("header_test", "header_value")); + assertAcked(prepareCreate("test")); GetMappingsResponse response = client().admin().indices().prepareGetMappings("test").get(); assertThat(response, notNullValue()); @@ -70,7 +70,7 @@ public class IndexTemplateFilteringIT extends ESIntegTestCase { @Override public boolean apply(CreateIndexClusterStateUpdateRequest request, IndexTemplateMetaData template) { //make sure that no_match template is filtered out before the custom filters as it doesn't match the index name - return (template.name().equals("template2") || template.name().equals("no_match")) && request.originalMessage().getHeader("header_test").equals("header_value"); + return (template.name().equals("template2") || template.name().equals("no_match")); } } diff --git a/core/src/test/java/org/elasticsearch/plugins/responseheader/TestResponseHeaderRestAction.java b/core/src/test/java/org/elasticsearch/plugins/responseheader/TestResponseHeaderRestAction.java index 4b1645a4ec..39432bd01e 100644 --- a/core/src/test/java/org/elasticsearch/plugins/responseheader/TestResponseHeaderRestAction.java +++ b/core/src/test/java/org/elasticsearch/plugins/responseheader/TestResponseHeaderRestAction.java @@ -33,7 +33,7 @@ public class TestResponseHeaderRestAction extends BaseRestHandler { @Inject public TestResponseHeaderRestAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(RestRequest.Method.GET, "/_protected", this); } diff --git a/core/src/test/java/org/elasticsearch/rest/HeadersAndContextCopyClientTests.java b/core/src/test/java/org/elasticsearch/rest/HeadersAndContextCopyClientTests.java deleted file mode 100644 index 238e16d4b4..0000000000 --- a/core/src/test/java/org/elasticsearch/rest/HeadersAndContextCopyClientTests.java +++ /dev/null @@ -1,425 +0,0 @@ -/* - * Licensed to Elasticsearch under one or more contributor - * license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright - * ownership. Elasticsearch licenses this file to you under - * the Apache License, Version 2.0 (the "License"); you may - * not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.elasticsearch.rest; - -import org.elasticsearch.action.ActionRequest; -import org.elasticsearch.action.ActionRequestBuilder; -import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest; -import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest; -import org.elasticsearch.action.admin.cluster.stats.ClusterStatsRequest; -import org.elasticsearch.action.admin.indices.close.CloseIndexRequest; -import org.elasticsearch.action.admin.indices.create.CreateIndexRequest; -import org.elasticsearch.action.admin.indices.flush.FlushRequest; -import org.elasticsearch.action.get.GetRequest; -import org.elasticsearch.action.index.IndexRequest; -import org.elasticsearch.action.search.SearchRequest; -import org.elasticsearch.client.Client; -import org.elasticsearch.client.Requests; -import org.elasticsearch.common.collect.ImmutableOpenMap; -import org.elasticsearch.common.settings.Settings; -import org.elasticsearch.test.ESTestCase; -import org.elasticsearch.test.rest.FakeRestRequest; - -import java.util.Arrays; -import java.util.HashMap; -import java.util.HashSet; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.TimeUnit; - -import static org.hamcrest.CoreMatchers.equalTo; -import static org.hamcrest.CoreMatchers.notNullValue; -import static org.hamcrest.Matchers.is; - -public class HeadersAndContextCopyClientTests extends ESTestCase { - - public void testRegisterRelevantHeaders() throws InterruptedException { - - final RestController restController = new RestController(Settings.EMPTY); - - int iterations = randomIntBetween(1, 5); - - Set<String> headers = new HashSet<>(); - ExecutorService executorService = Executors.newFixedThreadPool(iterations); - for (int i = 0; i < iterations; i++) { - int headersCount = randomInt(10); - final Set<String> newHeaders = new HashSet<>(); - for (int j = 0; j < headersCount; j++) { - String usefulHeader = randomRealisticUnicodeOfLengthBetween(1, 30); - newHeaders.add(usefulHeader); - } - headers.addAll(newHeaders); - - executorService.submit(new Runnable() { - @Override - public void run() { - restController.registerRelevantHeaders(newHeaders.toArray(new String[newHeaders.size()])); - } - }); - } - - executorService.shutdown(); - assertThat(executorService.awaitTermination(1, TimeUnit.SECONDS), equalTo(true)); - String[] relevantHeaders = restController.relevantHeaders().toArray(new String[restController.relevantHeaders().size()]); - assertThat(relevantHeaders.length, equalTo(headers.size())); - - Arrays.sort(relevantHeaders); - String[] headersArray = new String[headers.size()]; - headersArray = headers.toArray(headersArray); - Arrays.sort(headersArray); - assertThat(relevantHeaders, equalTo(headersArray)); - } - - public void testCopyHeadersRequest() { - Map<String, String> transportHeaders = randomHeaders(randomIntBetween(0, 10)); - Map<String, String> restHeaders = randomHeaders(randomIntBetween(0, 10)); - Map<String, String> copiedHeaders = randomHeadersFrom(restHeaders); - Set<String> usefulRestHeaders = new HashSet<>(copiedHeaders.keySet()); - usefulRestHeaders.addAll(randomMap(randomIntBetween(0, 10), "useful-").keySet()); - Map<String, String> restContext = randomContext(randomIntBetween(0, 10)); - Map<String, String> transportContext = onlyOnLeft(randomContext(randomIntBetween(0, 10)), restContext); - - Map<String, String> expectedHeaders = new HashMap<>(); - expectedHeaders.putAll(transportHeaders); - expectedHeaders.putAll(copiedHeaders); - - Map<String, String> expectedContext = new HashMap<>(); - expectedContext.putAll(transportContext); - expectedContext.putAll(restContext); - - try (Client client = client(new NoOpClient(getTestName()), new FakeRestRequest(restHeaders, restContext), usefulRestHeaders)) { - - SearchRequest searchRequest = Requests.searchRequest(); - putHeaders(searchRequest, transportHeaders); - putContext(searchRequest, transportContext); - assertHeaders(searchRequest, transportHeaders); - client.search(searchRequest); - assertHeaders(searchRequest, expectedHeaders); - assertContext(searchRequest, expectedContext); - - GetRequest getRequest = Requests.getRequest("index"); - putHeaders(getRequest, transportHeaders); - putContext(getRequest, transportContext); - assertHeaders(getRequest, transportHeaders); - client.get(getRequest); - assertHeaders(getRequest, expectedHeaders); - assertContext(getRequest, expectedContext); - - IndexRequest indexRequest = Requests.indexRequest(); - putHeaders(indexRequest, transportHeaders); - putContext(indexRequest, transportContext); - assertHeaders(indexRequest, transportHeaders); - client.index(indexRequest); - assertHeaders(indexRequest, expectedHeaders); - assertContext(indexRequest, expectedContext); - } - } - - public void testCopyHeadersClusterAdminRequest() { - Map<String, String> transportHeaders = randomHeaders(randomIntBetween(0, 10)); - Map<String, String> restHeaders = randomHeaders(randomIntBetween(0, 10)); - Map<String, String> copiedHeaders = randomHeadersFrom(restHeaders); - Set<String> usefulRestHeaders = new HashSet<>(copiedHeaders.keySet()); - usefulRestHeaders.addAll(randomMap(randomIntBetween(0, 10), "useful-").keySet()); - Map<String, String> restContext = randomContext(randomIntBetween(0, 10)); - Map<String, String> transportContext = onlyOnLeft(randomContext(randomIntBetween(0, 10)), restContext); - - HashMap<String, String> expectedHeaders = new HashMap<>(); - expectedHeaders.putAll(transportHeaders); - expectedHeaders.putAll(copiedHeaders); - - Map<String, String> expectedContext = new HashMap<>(); - expectedContext.putAll(transportContext); - expectedContext.putAll(restContext); - - try (Client client = client(new NoOpClient(getTestName()), new FakeRestRequest(restHeaders, expectedContext), usefulRestHeaders)) { - - ClusterHealthRequest clusterHealthRequest = Requests.clusterHealthRequest(); - putHeaders(clusterHealthRequest, transportHeaders); - putContext(clusterHealthRequest, transportContext); - assertHeaders(clusterHealthRequest, transportHeaders); - client.admin().cluster().health(clusterHealthRequest); - assertHeaders(clusterHealthRequest, expectedHeaders); - assertContext(clusterHealthRequest, expectedContext); - - ClusterStateRequest clusterStateRequest = Requests.clusterStateRequest(); - putHeaders(clusterStateRequest, transportHeaders); - putContext(clusterStateRequest, transportContext); - assertHeaders(clusterStateRequest, transportHeaders); - client.admin().cluster().state(clusterStateRequest); - assertHeaders(clusterStateRequest, expectedHeaders); - assertContext(clusterStateRequest, expectedContext); - - ClusterStatsRequest clusterStatsRequest = Requests.clusterStatsRequest(); - putHeaders(clusterStatsRequest, transportHeaders); - putContext(clusterStatsRequest, transportContext); - assertHeaders(clusterStatsRequest, transportHeaders); - client.admin().cluster().clusterStats(clusterStatsRequest); - assertHeaders(clusterStatsRequest, expectedHeaders); - assertContext(clusterStatsRequest, expectedContext); - } - } - - public void testCopyHeadersIndicesAdminRequest() { - Map<String, String> transportHeaders = randomHeaders(randomIntBetween(0, 10)); - Map<String, String> restHeaders = randomHeaders(randomIntBetween(0, 10)); - Map<String, String> copiedHeaders = randomHeadersFrom(restHeaders); - Set<String> usefulRestHeaders = new HashSet<>(copiedHeaders.keySet()); - usefulRestHeaders.addAll(randomMap(randomIntBetween(0, 10), "useful-").keySet()); - Map<String, String> restContext = randomContext(randomIntBetween(0, 10)); - Map<String, String> transportContext = onlyOnLeft(randomContext(randomIntBetween(0, 10)), restContext); - - HashMap<String, String> expectedHeaders = new HashMap<>(); - expectedHeaders.putAll(transportHeaders); - expectedHeaders.putAll(copiedHeaders); - - Map<String, String> expectedContext = new HashMap<>(); - expectedContext.putAll(transportContext); - expectedContext.putAll(restContext); - - try (Client client = client(new NoOpClient(getTestName()), new FakeRestRequest(restHeaders, restContext), usefulRestHeaders)) { - - CreateIndexRequest createIndexRequest = Requests.createIndexRequest("test"); - putHeaders(createIndexRequest, transportHeaders); - putContext(createIndexRequest, transportContext); - assertHeaders(createIndexRequest, transportHeaders); - client.admin().indices().create(createIndexRequest); - assertHeaders(createIndexRequest, expectedHeaders); - assertContext(createIndexRequest, expectedContext); - - CloseIndexRequest closeIndexRequest = Requests.closeIndexRequest("test"); - putHeaders(closeIndexRequest, transportHeaders); - putContext(closeIndexRequest, transportContext); - assertHeaders(closeIndexRequest, transportHeaders); - client.admin().indices().close(closeIndexRequest); - assertHeaders(closeIndexRequest, expectedHeaders); - assertContext(closeIndexRequest, expectedContext); - - FlushRequest flushRequest = Requests.flushRequest(); - putHeaders(flushRequest, transportHeaders); - putContext(flushRequest, transportContext); - assertHeaders(flushRequest, transportHeaders); - client.admin().indices().flush(flushRequest); - assertHeaders(flushRequest, expectedHeaders); - assertContext(flushRequest, expectedContext); - } - } - - public void testCopyHeadersRequestBuilder() { - Map<String, String> transportHeaders = randomHeaders(randomIntBetween(0, 10)); - Map<String, String> restHeaders = randomHeaders(randomIntBetween(0, 10)); - Map<String, String> copiedHeaders = randomHeadersFrom(restHeaders); - Set<String> usefulRestHeaders = new HashSet<>(copiedHeaders.keySet()); - usefulRestHeaders.addAll(randomMap(randomIntBetween(0, 10), "useful-").keySet()); - Map<String, String> restContext = randomContext(randomIntBetween(0, 10)); - Map<String, String> transportContext = onlyOnLeft(randomContext(randomIntBetween(0, 10)), restContext); - - HashMap<String, String> expectedHeaders = new HashMap<>(); - expectedHeaders.putAll(transportHeaders); - expectedHeaders.putAll(copiedHeaders); - - Map<String, String> expectedContext = new HashMap<>(); - expectedContext.putAll(transportContext); - expectedContext.putAll(restContext); - - try (Client client = client(new NoOpClient(getTestName()), new FakeRestRequest(restHeaders, restContext), usefulRestHeaders)) { - - ActionRequestBuilder requestBuilders[] = new ActionRequestBuilder[]{ - client.prepareIndex("index", "type"), - client.prepareGet("index", "type", "id"), - client.prepareBulk(), - client.prepareDelete(), - client.prepareIndex(), - client.prepareClearScroll(), - client.prepareMultiGet(), - }; - - for (ActionRequestBuilder requestBuilder : requestBuilders) { - putHeaders(requestBuilder.request(), transportHeaders); - putContext(requestBuilder.request(), transportContext); - assertHeaders(requestBuilder.request(), transportHeaders); - requestBuilder.get(); - assertHeaders(requestBuilder.request(), expectedHeaders); - assertContext(requestBuilder.request(), expectedContext); - } - } - } - - public void testCopyHeadersClusterAdminRequestBuilder() { - Map<String, String> transportHeaders = randomHeaders(randomIntBetween(0, 10)); - Map<String, String> restHeaders = randomHeaders(randomIntBetween(0, 10)); - Map<String, String> copiedHeaders = randomHeadersFrom(restHeaders); - Set<String> usefulRestHeaders = new HashSet<>(copiedHeaders.keySet()); - usefulRestHeaders.addAll(randomMap(randomIntBetween(0, 10), "useful-").keySet()); - Map<String, String> restContext = randomContext(randomIntBetween(0, 10)); - Map<String, String> transportContext = onlyOnLeft(randomContext(randomIntBetween(0, 10)), restContext); - - HashMap<String, String> expectedHeaders = new HashMap<>(); - expectedHeaders.putAll(transportHeaders); - expectedHeaders.putAll(copiedHeaders); - - Map<String, String> expectedContext = new HashMap<>(); - expectedContext.putAll(transportContext); - expectedContext.putAll(restContext); - - try (Client client = client(new NoOpClient(getTestName()), new FakeRestRequest(restHeaders, restContext), usefulRestHeaders)) { - - ActionRequestBuilder requestBuilders[] = new ActionRequestBuilder[]{ - client.admin().cluster().prepareNodesInfo(), - client.admin().cluster().prepareClusterStats(), - client.admin().cluster().prepareState(), - client.admin().cluster().prepareCreateSnapshot("repo", "name"), - client.admin().cluster().prepareHealth(), - client.admin().cluster().prepareReroute() - }; - - for (ActionRequestBuilder requestBuilder : requestBuilders) { - putHeaders(requestBuilder.request(), transportHeaders); - putContext(requestBuilder.request(), transportContext); - assertHeaders(requestBuilder.request(), transportHeaders); - requestBuilder.get(); - assertHeaders(requestBuilder.request(), expectedHeaders); - assertContext(requestBuilder.request(), expectedContext); - } - } - } - - public void testCopyHeadersIndicesAdminRequestBuilder() { - Map<String, String> transportHeaders = randomHeaders(randomIntBetween(0, 10)); - Map<String, String> restHeaders = randomHeaders(randomIntBetween(0, 10)); - Map<String, String> copiedHeaders = randomHeadersFrom(restHeaders); - Set<String> usefulRestHeaders = new HashSet<>(copiedHeaders.keySet()); - usefulRestHeaders.addAll(randomMap(randomIntBetween(0, 10), "useful-").keySet()); - Map<String, String> restContext = randomContext(randomIntBetween(0, 10)); - Map<String, String> transportContext = onlyOnLeft(randomContext(randomIntBetween(0, 10)), restContext); - - HashMap<String, String> expectedHeaders = new HashMap<>(); - expectedHeaders.putAll(transportHeaders); - expectedHeaders.putAll(copiedHeaders); - - Map<String, String> expectedContext = new HashMap<>(); - expectedContext.putAll(transportContext); - expectedContext.putAll(restContext); - - try (Client client = client(new NoOpClient(getTestName()), new FakeRestRequest(restHeaders, restContext), usefulRestHeaders)) { - - ActionRequestBuilder requestBuilders[] = new ActionRequestBuilder[]{ - client.admin().indices().prepareValidateQuery(), - client.admin().indices().prepareCreate("test"), - client.admin().indices().prepareAliases(), - client.admin().indices().prepareAnalyze("text"), - client.admin().indices().prepareTypesExists("type"), - client.admin().indices().prepareClose() - }; - - for (ActionRequestBuilder requestBuilder : requestBuilders) { - putHeaders(requestBuilder.request(), transportHeaders); - putContext(requestBuilder.request(), transportContext); - assertHeaders(requestBuilder.request(), transportHeaders); - requestBuilder.get(); - assertHeaders(requestBuilder.request(), expectedHeaders); - assertContext(requestBuilder.request(), expectedContext); - } - } - } - - private static Map<String, String> randomHeaders(int count) { - return randomMap(count, "header-"); - } - - private static Map<String, String> randomContext(int count) { - return randomMap(count, "context-"); - } - - private static Map<String, String> randomMap(int count, String prefix) { - Map<String, String> headers = new HashMap<>(); - for (int i = 0; i < count; i++) { - headers.put(prefix + randomInt(30), randomAsciiOfLength(10)); - } - return headers; - } - - private static Map<String, String> randomHeadersFrom(Map<String, String> headers) { - Map<String, String> newHeaders = new HashMap<>(); - if (headers.isEmpty()) { - return newHeaders; - } - int i = randomInt(headers.size() - 1); - for (Map.Entry<String, String> entry : headers.entrySet()) { - if (randomInt(i) == 0) { - newHeaders.put(entry.getKey(), entry.getValue()); - } - } - return newHeaders; - } - - private static Client client(Client noOpClient, RestRequest restRequest, Set<String> usefulRestHeaders) { - return new BaseRestHandler.HeadersAndContextCopyClient(noOpClient, restRequest, usefulRestHeaders); - } - - private static void putHeaders(ActionRequest<?> request, Map<String, String> headers) { - for (Map.Entry<String, String> header : headers.entrySet()) { - request.putHeader(header.getKey(), header.getValue()); - } - } - - private static void putContext(ActionRequest<?> request, Map<String, String> context) { - for (Map.Entry<String, String> header : context.entrySet()) { - request.putInContext(header.getKey(), header.getValue()); - } - } - - private static void assertHeaders(ActionRequest<?> request, Map<String, String> headers) { - if (headers.size() == 0) { - assertThat(request.getHeaders() == null || request.getHeaders().size() == 0, equalTo(true)); - } else { - assertThat(request.getHeaders(), notNullValue()); - assertThat(request.getHeaders().size(), equalTo(headers.size())); - for (String key : request.getHeaders()) { - assertThat(headers.get(key), equalTo(request.getHeader(key))); - } - } - } - - private static void assertContext(ActionRequest<?> request, Map<String, String> context) { - if (context.size() == 0) { - assertThat(request.isContextEmpty(), is(true)); - } else { - ImmutableOpenMap map = request.getContext(); - assertThat(map, notNullValue()); - assertThat(map.size(), equalTo(context.size())); - for (Object key : map.keys()) { - assertThat(context.get(key), equalTo(request.getFromContext(key))); - } - } - } - - private static Map<String, String> onlyOnLeft(Map<String, String> left, Map<String, String> right) { - Map<String, String> map = new HashMap<>(); - for (Map.Entry<String, String> entry : left.entrySet()) { - if (!right.containsKey(entry.getKey())) { - map.put(entry.getKey(), entry.getValue()); - } - } - return map; - } -} diff --git a/core/src/test/java/org/elasticsearch/rest/NoOpClient.java b/core/src/test/java/org/elasticsearch/rest/NoOpClient.java index 245bdb96a3..4f2a1ab07b 100644 --- a/core/src/test/java/org/elasticsearch/rest/NoOpClient.java +++ b/core/src/test/java/org/elasticsearch/rest/NoOpClient.java @@ -26,7 +26,6 @@ import org.elasticsearch.action.ActionRequest; import org.elasticsearch.action.ActionRequestBuilder; import org.elasticsearch.action.ActionResponse; import org.elasticsearch.client.support.AbstractClient; -import org.elasticsearch.client.support.Headers; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.threadpool.ThreadPool; @@ -35,7 +34,7 @@ import java.util.concurrent.TimeUnit; public class NoOpClient extends AbstractClient { public NoOpClient(String testName) { - super(Settings.EMPTY, new ThreadPool(testName), Headers.EMPTY); + super(Settings.EMPTY, new ThreadPool(testName)); } @Override @@ -51,4 +50,4 @@ public class NoOpClient extends AbstractClient { throw new ElasticsearchException(t.getMessage(), t); } } -}
\ No newline at end of file +} diff --git a/core/src/test/java/org/elasticsearch/rest/RestControllerTests.java b/core/src/test/java/org/elasticsearch/rest/RestControllerTests.java new file mode 100644 index 0000000000..d6e1a97ac8 --- /dev/null +++ b/core/src/test/java/org/elasticsearch/rest/RestControllerTests.java @@ -0,0 +1,99 @@ +/* + * Licensed to Elasticsearch under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.elasticsearch.rest; + +import org.elasticsearch.common.settings.Settings; +import org.elasticsearch.common.util.concurrent.ThreadContext; +import org.elasticsearch.test.ESTestCase; +import org.elasticsearch.test.rest.FakeRestRequest; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import static org.hamcrest.CoreMatchers.equalTo; + +public class RestControllerTests extends ESTestCase { + + public void testRegisterRelevantHeaders() throws InterruptedException { + + final RestController restController = new RestController(Settings.EMPTY); + + int iterations = randomIntBetween(1, 5); + + Set<String> headers = new HashSet<>(); + ExecutorService executorService = Executors.newFixedThreadPool(iterations); + for (int i = 0; i < iterations; i++) { + int headersCount = randomInt(10); + final Set<String> newHeaders = new HashSet<>(); + for (int j = 0; j < headersCount; j++) { + String usefulHeader = randomRealisticUnicodeOfLengthBetween(1, 30); + newHeaders.add(usefulHeader); + } + headers.addAll(newHeaders); + + executorService.submit((Runnable) () -> restController.registerRelevantHeaders(newHeaders.toArray(new String[newHeaders.size()]))); + } + + executorService.shutdown(); + assertThat(executorService.awaitTermination(1, TimeUnit.SECONDS), equalTo(true)); + String[] relevantHeaders = restController.relevantHeaders().toArray(new String[restController.relevantHeaders().size()]); + assertThat(relevantHeaders.length, equalTo(headers.size())); + + Arrays.sort(relevantHeaders); + String[] headersArray = new String[headers.size()]; + headersArray = headers.toArray(headersArray); + Arrays.sort(headersArray); + assertThat(relevantHeaders, equalTo(headersArray)); + } + + public void testApplyRelevantHeaders() { + final ThreadContext threadContext = new ThreadContext(Settings.EMPTY); + final RestController restController = new RestController(Settings.EMPTY) { + @Override + boolean checkRequestParameters(RestRequest request, RestChannel channel) { + return true; + } + + @Override + void executeHandler(RestRequest request, RestChannel channel) throws Exception { + assertEquals("true", threadContext.getHeader("header.1")); + assertEquals("true", threadContext.getHeader("header.2")); + assertNull(threadContext.getHeader("header.3")); + + } + }; + threadContext.putHeader("header.3", "true"); + restController.registerRelevantHeaders("header.1", "header.2"); + Map<String, String> restHeaders = new HashMap<>(); + restHeaders.put("header.1", "true"); + restHeaders.put("header.2", "true"); + restHeaders.put("header.3", "false"); + restController.dispatchRequest(new FakeRestRequest(restHeaders), null, threadContext); + assertNull(threadContext.getHeader("header.1")); + assertNull(threadContext.getHeader("header.2")); + assertEquals("true", threadContext.getHeader("header.3")); + } +} diff --git a/core/src/test/java/org/elasticsearch/rest/RestFilterChainTests.java b/core/src/test/java/org/elasticsearch/rest/RestFilterChainTests.java index b66d00cd6a..56ae8e2183 100644 --- a/core/src/test/java/org/elasticsearch/rest/RestFilterChainTests.java +++ b/core/src/test/java/org/elasticsearch/rest/RestFilterChainTests.java @@ -23,6 +23,7 @@ import org.elasticsearch.common.Nullable; import org.elasticsearch.common.bytes.BytesReference; import org.elasticsearch.common.io.stream.BytesStreamOutput; import org.elasticsearch.common.settings.Settings; +import org.elasticsearch.common.util.concurrent.ThreadContext; import org.elasticsearch.common.xcontent.XContentBuilder; import org.elasticsearch.test.ESTestCase; import org.elasticsearch.test.rest.FakeRestRequest; @@ -84,7 +85,7 @@ public class RestFilterChainTests extends ESTestCase { FakeRestRequest fakeRestRequest = new FakeRestRequest(); FakeRestChannel fakeRestChannel = new FakeRestChannel(fakeRestRequest, 1); - restController.dispatchRequest(fakeRestRequest, fakeRestChannel); + restController.dispatchRequest(fakeRestRequest, fakeRestChannel, new ThreadContext(Settings.EMPTY)); assertThat(fakeRestChannel.await(), equalTo(true)); @@ -142,7 +143,7 @@ public class RestFilterChainTests extends ESTestCase { FakeRestRequest fakeRestRequest = new FakeRestRequest(); FakeRestChannel fakeRestChannel = new FakeRestChannel(fakeRestRequest, additionalContinueCount + 1); - restController.dispatchRequest(fakeRestRequest, fakeRestChannel); + restController.dispatchRequest(fakeRestRequest, fakeRestChannel, new ThreadContext(Settings.EMPTY)); fakeRestChannel.await(); assertThat(testFilter.runs.get(), equalTo(1)); diff --git a/core/src/test/java/org/elasticsearch/rest/RestRequestTests.java b/core/src/test/java/org/elasticsearch/rest/RestRequestTests.java deleted file mode 100644 index 8e60b28f37..0000000000 --- a/core/src/test/java/org/elasticsearch/rest/RestRequestTests.java +++ /dev/null @@ -1,107 +0,0 @@ -/* - * Licensed to Elasticsearch under one or more contributor - * license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright - * ownership. Elasticsearch licenses this file to you under - * the Apache License, Version 2.0 (the "License"); you may - * not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.elasticsearch.rest; - -import org.elasticsearch.common.bytes.BytesReference; -import org.elasticsearch.common.collect.ImmutableOpenMap; -import org.elasticsearch.test.ESTestCase; - -import java.util.Map; - -import static org.hamcrest.Matchers.equalTo; -import static org.hamcrest.Matchers.is; - -/** - * - */ -public class RestRequestTests extends ESTestCase { - public void testContext() throws Exception { - int count = randomInt(10); - Request request = new Request(); - for (int i = 0; i < count; i++) { - request.putInContext("key" + i, "val" + i); - } - assertThat(request.isContextEmpty(), is(count == 0)); - assertThat(request.contextSize(), is(count)); - ImmutableOpenMap<Object, Object> ctx = request.getContext(); - for (int i = 0; i < count; i++) { - assertThat(request.hasInContext("key" + i), is(true)); - assertThat((String) request.getFromContext("key" + i), equalTo("val" + i)); - assertThat((String) ctx.get("key" + i), equalTo("val" + i)); - } - } - - public static class Request extends RestRequest { - @Override - public Method method() { - return null; - } - - @Override - public String uri() { - return null; - } - - @Override - public String rawPath() { - return null; - } - - @Override - public boolean hasContent() { - return false; - } - - @Override - public BytesReference content() { - return null; - } - - @Override - public String header(String name) { - return null; - } - - @Override - public Iterable<Map.Entry<String, String>> headers() { - return null; - } - - @Override - public boolean hasParam(String key) { - return false; - } - - @Override - public String param(String key) { - return null; - } - - @Override - public Map<String, String> params() { - return null; - } - - @Override - public String param(String key, String defaultValue) { - return null; - } - } -} diff --git a/core/src/test/java/org/elasticsearch/script/FileScriptTests.java b/core/src/test/java/org/elasticsearch/script/FileScriptTests.java index 987aef90bc..7b891777b3 100644 --- a/core/src/test/java/org/elasticsearch/script/FileScriptTests.java +++ b/core/src/test/java/org/elasticsearch/script/FileScriptTests.java @@ -18,7 +18,6 @@ */ package org.elasticsearch.script; -import org.elasticsearch.common.ContextAndHeaderHolder; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.env.Environment; import org.elasticsearch.test.ESTestCase; @@ -49,16 +48,14 @@ public class FileScriptTests extends ESTestCase { } public void testFileScriptFound() throws Exception { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); Settings settings = Settings.builder() .put("script.engine." + MockScriptEngine.NAME + ".file.aggs", false).build(); ScriptService scriptService = makeScriptService(settings); Script script = new Script("script1", ScriptService.ScriptType.FILE, MockScriptEngine.NAME, null); - assertNotNull(scriptService.compile(script, ScriptContext.Standard.SEARCH, contextAndHeaders, Collections.emptyMap())); + assertNotNull(scriptService.compile(script, ScriptContext.Standard.SEARCH, Collections.emptyMap())); } public void testAllOpsDisabled() throws Exception { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); Settings settings = Settings.builder() .put("script.engine." + MockScriptEngine.NAME + ".file.aggs", false) .put("script.engine." + MockScriptEngine.NAME + ".file.search", false) @@ -68,7 +65,7 @@ public class FileScriptTests extends ESTestCase { Script script = new Script("script1", ScriptService.ScriptType.FILE, MockScriptEngine.NAME, null); for (ScriptContext context : ScriptContext.Standard.values()) { try { - scriptService.compile(script, context, contextAndHeaders, Collections.emptyMap()); + scriptService.compile(script, context, Collections.emptyMap()); fail(context.getKey() + " script should have been rejected"); } catch(Exception e) { assertTrue(e.getMessage(), e.getMessage().contains("scripts of type [file], operation [" + context.getKey() + "] and lang [" + MockScriptEngine.NAME + "] are disabled")); diff --git a/core/src/test/java/org/elasticsearch/script/NativeScriptTests.java b/core/src/test/java/org/elasticsearch/script/NativeScriptTests.java index 47adeabe02..78314c7a7b 100644 --- a/core/src/test/java/org/elasticsearch/script/NativeScriptTests.java +++ b/core/src/test/java/org/elasticsearch/script/NativeScriptTests.java @@ -19,7 +19,6 @@ package org.elasticsearch.script; -import org.elasticsearch.common.ContextAndHeaderHolder; import org.elasticsearch.common.Nullable; import org.elasticsearch.common.inject.Injector; import org.elasticsearch.common.inject.ModulesBuilder; @@ -47,7 +46,6 @@ import static org.hamcrest.Matchers.notNullValue; public class NativeScriptTests extends ESTestCase { public void testNativeScript() throws InterruptedException { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); Settings settings = Settings.settingsBuilder() .put("name", "testNativeScript") .put("path.home", createTempDir()) @@ -63,13 +61,12 @@ public class NativeScriptTests extends ESTestCase { ScriptService scriptService = injector.getInstance(ScriptService.class); ExecutableScript executable = scriptService.executable(new Script("my", ScriptType.INLINE, NativeScriptEngineService.NAME, null), - ScriptContext.Standard.SEARCH, contextAndHeaders, Collections.emptyMap()); + ScriptContext.Standard.SEARCH, Collections.emptyMap()); assertThat(executable.run().toString(), equalTo("test")); terminate(injector.getInstance(ThreadPool.class)); } public void testFineGrainedSettingsDontAffectNativeScripts() throws IOException { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); Settings.Builder builder = Settings.settingsBuilder(); if (randomBoolean()) { ScriptType scriptType = randomFrom(ScriptType.values()); @@ -89,7 +86,7 @@ public class NativeScriptTests extends ESTestCase { for (ScriptContext scriptContext : scriptContextRegistry.scriptContexts()) { assertThat(scriptService.compile(new Script("my", ScriptType.INLINE, NativeScriptEngineService.NAME, null), scriptContext, - contextAndHeaders, Collections.emptyMap()), notNullValue()); + Collections.emptyMap()), notNullValue()); } } diff --git a/core/src/test/java/org/elasticsearch/script/ScriptContextTests.java b/core/src/test/java/org/elasticsearch/script/ScriptContextTests.java index 019eb7c74a..42378cb702 100644 --- a/core/src/test/java/org/elasticsearch/script/ScriptContextTests.java +++ b/core/src/test/java/org/elasticsearch/script/ScriptContextTests.java @@ -19,7 +19,6 @@ package org.elasticsearch.script; -import org.elasticsearch.common.ContextAndHeaderHolder; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.env.Environment; import org.elasticsearch.test.ESTestCase; @@ -53,12 +52,11 @@ public class ScriptContextTests extends ESTestCase { } public void testCustomGlobalScriptContextSettings() throws Exception { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); ScriptService scriptService = makeScriptService(); for (ScriptService.ScriptType scriptType : ScriptService.ScriptType.values()) { try { Script script = new Script("1", scriptType, MockScriptEngine.NAME, null); - scriptService.compile(script, new ScriptContext.Plugin(PLUGIN_NAME, "custom_globally_disabled_op"), contextAndHeaders, Collections.emptyMap()); + scriptService.compile(script, new ScriptContext.Plugin(PLUGIN_NAME, "custom_globally_disabled_op"), Collections.emptyMap()); fail("script compilation should have been rejected"); } catch (ScriptException e) { assertThat(e.getMessage(), containsString("scripts of type [" + scriptType + "], operation [" + PLUGIN_NAME + "_custom_globally_disabled_op] and lang [" + MockScriptEngine.NAME + "] are disabled")); @@ -67,29 +65,27 @@ public class ScriptContextTests extends ESTestCase { } public void testCustomScriptContextSettings() throws Exception { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); ScriptService scriptService = makeScriptService(); Script script = new Script("1", ScriptService.ScriptType.INLINE, MockScriptEngine.NAME, null); try { - scriptService.compile(script, new ScriptContext.Plugin(PLUGIN_NAME, "custom_exp_disabled_op"), contextAndHeaders, Collections.emptyMap()); + scriptService.compile(script, new ScriptContext.Plugin(PLUGIN_NAME, "custom_exp_disabled_op"), Collections.emptyMap()); fail("script compilation should have been rejected"); } catch (ScriptException e) { assertTrue(e.getMessage(), e.getMessage().contains("scripts of type [inline], operation [" + PLUGIN_NAME + "_custom_exp_disabled_op] and lang [" + MockScriptEngine.NAME + "] are disabled")); } // still works for other script contexts - assertNotNull(scriptService.compile(script, ScriptContext.Standard.AGGS, contextAndHeaders, Collections.emptyMap())); - assertNotNull(scriptService.compile(script, ScriptContext.Standard.SEARCH, contextAndHeaders, Collections.emptyMap())); - assertNotNull(scriptService.compile(script, new ScriptContext.Plugin(PLUGIN_NAME, "custom_op"), contextAndHeaders, Collections.emptyMap())); + assertNotNull(scriptService.compile(script, ScriptContext.Standard.AGGS, Collections.emptyMap())); + assertNotNull(scriptService.compile(script, ScriptContext.Standard.SEARCH, Collections.emptyMap())); + assertNotNull(scriptService.compile(script, new ScriptContext.Plugin(PLUGIN_NAME, "custom_op"), Collections.emptyMap())); } public void testUnknownPluginScriptContext() throws Exception { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); ScriptService scriptService = makeScriptService(); for (ScriptService.ScriptType scriptType : ScriptService.ScriptType.values()) { try { Script script = new Script("1", scriptType, MockScriptEngine.NAME, null); - scriptService.compile(script, new ScriptContext.Plugin(PLUGIN_NAME, "unknown"), contextAndHeaders, Collections.emptyMap()); + scriptService.compile(script, new ScriptContext.Plugin(PLUGIN_NAME, "unknown"), Collections.emptyMap()); fail("script compilation should have been rejected"); } catch (IllegalArgumentException e) { assertTrue(e.getMessage(), e.getMessage().contains("script context [" + PLUGIN_NAME + "_unknown] not supported")); @@ -98,7 +94,6 @@ public class ScriptContextTests extends ESTestCase { } public void testUnknownCustomScriptContext() throws Exception { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); ScriptContext context = new ScriptContext() { @Override public String getKey() { @@ -109,7 +104,7 @@ public class ScriptContextTests extends ESTestCase { for (ScriptService.ScriptType scriptType : ScriptService.ScriptType.values()) { try { Script script = new Script("1", scriptType, MockScriptEngine.NAME, null); - scriptService.compile(script, context, contextAndHeaders, Collections.emptyMap()); + scriptService.compile(script, context, Collections.emptyMap()); fail("script compilation should have been rejected"); } catch (IllegalArgumentException e) { assertTrue(e.getMessage(), e.getMessage().contains("script context [test] not supported")); diff --git a/core/src/test/java/org/elasticsearch/script/ScriptServiceTests.java b/core/src/test/java/org/elasticsearch/script/ScriptServiceTests.java index 3c939e7e91..f94835e90a 100644 --- a/core/src/test/java/org/elasticsearch/script/ScriptServiceTests.java +++ b/core/src/test/java/org/elasticsearch/script/ScriptServiceTests.java @@ -18,8 +18,6 @@ */ package org.elasticsearch.script; -import org.elasticsearch.common.ContextAndHeaderHolder; -import org.elasticsearch.common.HasContextAndHeaders; import org.elasticsearch.common.Nullable; import org.elasticsearch.common.io.Streams; import org.elasticsearch.common.settings.Settings; @@ -101,7 +99,7 @@ public class ScriptServiceTests extends ESTestCase { Environment environment = new Environment(finalSettings); scriptService = new ScriptService(finalSettings, environment, Collections.singleton(scriptEngineService), resourceWatcherService, scriptContextRegistry) { @Override - String getScriptFromIndex(String scriptLang, String id, HasContextAndHeaders headersContext) { + String getScriptFromIndex(String scriptLang, String id) { //mock the script that gets retrieved from an index return "100"; } @@ -119,7 +117,6 @@ public class ScriptServiceTests extends ESTestCase { public void testScriptsWithoutExtensions() throws IOException { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); buildScriptService(Settings.EMPTY); logger.info("--> setup two test files one with extension and another without"); Path testFileNoExt = scriptsFilePath.resolve("test_no_ext"); @@ -130,7 +127,7 @@ public class ScriptServiceTests extends ESTestCase { logger.info("--> verify that file with extension was correctly processed"); CompiledScript compiledScript = scriptService.compile(new Script("test_script", ScriptType.FILE, "test", null), - ScriptContext.Standard.SEARCH, contextAndHeaders, Collections.emptyMap()); + ScriptContext.Standard.SEARCH, Collections.emptyMap()); assertThat(compiledScript.compiled(), equalTo((Object) "compiled_test_file")); logger.info("--> delete both files"); @@ -141,7 +138,7 @@ public class ScriptServiceTests extends ESTestCase { logger.info("--> verify that file with extension was correctly removed"); try { scriptService.compile(new Script("test_script", ScriptType.FILE, "test", null), ScriptContext.Standard.SEARCH, - contextAndHeaders, Collections.emptyMap()); + Collections.emptyMap()); fail("the script test_script should no longer exist"); } catch (IllegalArgumentException ex) { assertThat(ex.getMessage(), containsString("Unable to find on disk file script [test_script] using lang [test]")); @@ -149,38 +146,34 @@ public class ScriptServiceTests extends ESTestCase { } public void testInlineScriptCompiledOnceCache() throws IOException { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); buildScriptService(Settings.EMPTY); CompiledScript compiledScript1 = scriptService.compile(new Script("1+1", ScriptType.INLINE, "test", null), - randomFrom(scriptContexts), contextAndHeaders, Collections.emptyMap()); + randomFrom(scriptContexts), Collections.emptyMap()); CompiledScript compiledScript2 = scriptService.compile(new Script("1+1", ScriptType.INLINE, "test", null), - randomFrom(scriptContexts), contextAndHeaders, Collections.emptyMap()); + randomFrom(scriptContexts), Collections.emptyMap()); assertThat(compiledScript1.compiled(), sameInstance(compiledScript2.compiled())); } public void testInlineScriptCompiledOnceMultipleLangAcronyms() throws IOException { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); buildScriptService(Settings.EMPTY); CompiledScript compiledScript1 = scriptService.compile(new Script("script", ScriptType.INLINE, "test", null), - randomFrom(scriptContexts), contextAndHeaders, Collections.emptyMap()); + randomFrom(scriptContexts), Collections.emptyMap()); CompiledScript compiledScript2 = scriptService.compile(new Script("script", ScriptType.INLINE, "test2", null), - randomFrom(scriptContexts), contextAndHeaders, Collections.emptyMap()); + randomFrom(scriptContexts), Collections.emptyMap()); assertThat(compiledScript1.compiled(), sameInstance(compiledScript2.compiled())); } public void testFileScriptCompiledOnceMultipleLangAcronyms() throws IOException { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); buildScriptService(Settings.EMPTY); createFileScripts("test"); CompiledScript compiledScript1 = scriptService.compile(new Script("file_script", ScriptType.FILE, "test", null), - randomFrom(scriptContexts), contextAndHeaders, Collections.emptyMap()); + randomFrom(scriptContexts), Collections.emptyMap()); CompiledScript compiledScript2 = scriptService.compile(new Script("file_script", ScriptType.FILE, "test2", null), - randomFrom(scriptContexts), contextAndHeaders, Collections.emptyMap()); + randomFrom(scriptContexts), Collections.emptyMap()); assertThat(compiledScript1.compiled(), sameInstance(compiledScript2.compiled())); } public void testDefaultBehaviourFineGrainedSettings() throws IOException { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); Settings.Builder builder = Settings.builder(); //rarely inject the default settings, which have no effect if (rarely()) { @@ -197,14 +190,13 @@ public class ScriptServiceTests extends ESTestCase { for (ScriptContext scriptContext : scriptContexts) { //custom engine is sandboxed, all scripts are enabled by default - assertCompileAccepted("test", "script", ScriptType.INLINE, scriptContext, contextAndHeaders); - assertCompileAccepted("test", "script", ScriptType.INDEXED, scriptContext, contextAndHeaders); - assertCompileAccepted("test", "file_script", ScriptType.FILE, scriptContext, contextAndHeaders); + assertCompileAccepted("test", "script", ScriptType.INLINE, scriptContext); + assertCompileAccepted("test", "script", ScriptType.INDEXED, scriptContext); + assertCompileAccepted("test", "file_script", ScriptType.FILE, scriptContext); } } public void testFineGrainedSettings() throws IOException { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); //collect the fine-grained settings to set for this run int numScriptSettings = randomIntBetween(0, ScriptType.values().length); Map<ScriptType, ScriptMode> scriptSourceSettings = new HashMap<>(); @@ -305,16 +297,16 @@ public class ScriptServiceTests extends ESTestCase { for (String lang : scriptEngineService.types()) { switch (scriptMode) { case ON: - assertCompileAccepted(lang, script, scriptType, scriptContext, contextAndHeaders); + assertCompileAccepted(lang, script, scriptType, scriptContext); break; case OFF: - assertCompileRejected(lang, script, scriptType, scriptContext, contextAndHeaders); + assertCompileRejected(lang, script, scriptType, scriptContext); break; case SANDBOX: if (scriptEngineService.sandboxed()) { - assertCompileAccepted(lang, script, scriptType, scriptContext, contextAndHeaders); + assertCompileAccepted(lang, script, scriptType, scriptContext); } else { - assertCompileRejected(lang, script, scriptType, scriptContext, contextAndHeaders); + assertCompileRejected(lang, script, scriptType, scriptContext); } break; } @@ -324,7 +316,6 @@ public class ScriptServiceTests extends ESTestCase { } public void testCompileNonRegisteredContext() throws IOException { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); buildScriptService(Settings.EMPTY); String pluginName; String unknownContext; @@ -336,7 +327,7 @@ public class ScriptServiceTests extends ESTestCase { for (String type : scriptEngineService.types()) { try { scriptService.compile(new Script("test", randomFrom(ScriptType.values()), type, null), new ScriptContext.Plugin( - pluginName, unknownContext), contextAndHeaders, Collections.emptyMap()); + pluginName, unknownContext), Collections.emptyMap()); fail("script compilation should have been rejected"); } catch(IllegalArgumentException e) { assertThat(e.getMessage(), containsString("script context [" + pluginName + "_" + unknownContext + "] not supported")); @@ -345,16 +336,14 @@ public class ScriptServiceTests extends ESTestCase { } public void testCompileCountedInCompilationStats() throws IOException { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); buildScriptService(Settings.EMPTY); - scriptService.compile(new Script("1+1", ScriptType.INLINE, "test", null), randomFrom(scriptContexts), contextAndHeaders, Collections.emptyMap()); + scriptService.compile(new Script("1+1", ScriptType.INLINE, "test", null), randomFrom(scriptContexts), Collections.emptyMap()); assertEquals(1L, scriptService.stats().getCompilations()); } public void testExecutableCountedInCompilationStats() throws IOException { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); buildScriptService(Settings.EMPTY); - scriptService.executable(new Script("1+1", ScriptType.INLINE, "test", null), randomFrom(scriptContexts), contextAndHeaders, Collections.emptyMap()); + scriptService.executable(new Script("1+1", ScriptType.INLINE, "test", null), randomFrom(scriptContexts), Collections.emptyMap()); assertEquals(1L, scriptService.stats().getCompilations()); } @@ -365,48 +354,43 @@ public class ScriptServiceTests extends ESTestCase { } public void testMultipleCompilationsCountedInCompilationStats() throws IOException { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); buildScriptService(Settings.EMPTY); int numberOfCompilations = randomIntBetween(1, 1024); for (int i = 0; i < numberOfCompilations; i++) { scriptService - .compile(new Script(i + " + " + i, ScriptType.INLINE, "test", null), randomFrom(scriptContexts), contextAndHeaders, Collections.emptyMap()); + .compile(new Script(i + " + " + i, ScriptType.INLINE, "test", null), randomFrom(scriptContexts), Collections.emptyMap()); } assertEquals(numberOfCompilations, scriptService.stats().getCompilations()); } public void testCompilationStatsOnCacheHit() throws IOException { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); Settings.Builder builder = Settings.builder(); builder.put(ScriptService.SCRIPT_CACHE_SIZE_SETTING, 1); buildScriptService(builder.build()); - scriptService.executable(new Script("1+1", ScriptType.INLINE, "test", null), randomFrom(scriptContexts), contextAndHeaders, Collections.emptyMap()); - scriptService.executable(new Script("1+1", ScriptType.INLINE, "test", null), randomFrom(scriptContexts), contextAndHeaders, Collections.emptyMap()); + scriptService.executable(new Script("1+1", ScriptType.INLINE, "test", null), randomFrom(scriptContexts), Collections.emptyMap()); + scriptService.executable(new Script("1+1", ScriptType.INLINE, "test", null), randomFrom(scriptContexts), Collections.emptyMap()); assertEquals(1L, scriptService.stats().getCompilations()); } public void testFileScriptCountedInCompilationStats() throws IOException { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); buildScriptService(Settings.EMPTY); createFileScripts("test"); - scriptService.compile(new Script("file_script", ScriptType.FILE, "test", null), randomFrom(scriptContexts), contextAndHeaders, Collections.emptyMap()); + scriptService.compile(new Script("file_script", ScriptType.FILE, "test", null), randomFrom(scriptContexts), Collections.emptyMap()); assertEquals(1L, scriptService.stats().getCompilations()); } public void testIndexedScriptCountedInCompilationStats() throws IOException { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); buildScriptService(Settings.EMPTY); - scriptService.compile(new Script("script", ScriptType.INDEXED, "test", null), randomFrom(scriptContexts), contextAndHeaders, Collections.emptyMap()); + scriptService.compile(new Script("script", ScriptType.INDEXED, "test", null), randomFrom(scriptContexts), Collections.emptyMap()); assertEquals(1L, scriptService.stats().getCompilations()); } public void testCacheEvictionCountedInCacheEvictionsStats() throws IOException { - ContextAndHeaderHolder contextAndHeaders = new ContextAndHeaderHolder(); Settings.Builder builder = Settings.builder(); builder.put(ScriptService.SCRIPT_CACHE_SIZE_SETTING, 1); buildScriptService(builder.build()); - scriptService.executable(new Script("1+1", ScriptType.INLINE, "test", null), randomFrom(scriptContexts), contextAndHeaders, Collections.emptyMap()); - scriptService.executable(new Script("2+2", ScriptType.INLINE, "test", null), randomFrom(scriptContexts), contextAndHeaders, Collections.emptyMap()); + scriptService.executable(new Script("1+1", ScriptType.INLINE, "test", null), randomFrom(scriptContexts), Collections.emptyMap()); + scriptService.executable(new Script("2+2", ScriptType.INLINE, "test", null), randomFrom(scriptContexts), Collections.emptyMap()); assertEquals(2L, scriptService.stats().getCompilations()); assertEquals(1L, scriptService.stats().getCacheEvictions()); } @@ -419,19 +403,17 @@ public class ScriptServiceTests extends ESTestCase { resourceWatcherService.notifyNow(); } - private void assertCompileRejected(String lang, String script, ScriptType scriptType, ScriptContext scriptContext, - HasContextAndHeaders contextAndHeaders) { + private void assertCompileRejected(String lang, String script, ScriptType scriptType, ScriptContext scriptContext) { try { - scriptService.compile(new Script(script, scriptType, lang, null), scriptContext, contextAndHeaders, Collections.emptyMap()); + scriptService.compile(new Script(script, scriptType, lang, null), scriptContext, Collections.emptyMap()); fail("compile should have been rejected for lang [" + lang + "], script_type [" + scriptType + "], scripted_op [" + scriptContext + "]"); } catch(ScriptException e) { //all good } } - private void assertCompileAccepted(String lang, String script, ScriptType scriptType, ScriptContext scriptContext, - HasContextAndHeaders contextAndHeaders) { - assertThat(scriptService.compile(new Script(script, scriptType, lang, null), scriptContext, contextAndHeaders, Collections.emptyMap()), notNullValue()); + private void assertCompileAccepted(String lang, String script, ScriptType scriptType, ScriptContext scriptContext) { + assertThat(scriptService.compile(new Script(script, scriptType, lang, null), scriptContext, Collections.emptyMap()), notNullValue()); } public static class TestEngineService implements ScriptEngineService { diff --git a/core/src/test/java/org/elasticsearch/search/suggest/CustomSuggester.java b/core/src/test/java/org/elasticsearch/search/suggest/CustomSuggester.java index 419316b526..35d495272c 100644 --- a/core/src/test/java/org/elasticsearch/search/suggest/CustomSuggester.java +++ b/core/src/test/java/org/elasticsearch/search/suggest/CustomSuggester.java @@ -54,7 +54,7 @@ public class CustomSuggester extends Suggester<CustomSuggester.CustomSuggestions @Override public SuggestContextParser getContextParser() { - return (parser, mapperService, fieldData, headersContext) -> { + return (parser, mapperService, fieldData) -> { Map<String, Object> options = parser.map(); CustomSuggestionsContext suggestionContext = new CustomSuggestionsContext(CustomSuggester.this, options); suggestionContext.setField((String) options.get("field")); diff --git a/core/src/test/java/org/elasticsearch/transport/AbstractSimpleTransportTestCase.java b/core/src/test/java/org/elasticsearch/transport/AbstractSimpleTransportTestCase.java index a5b6e08de3..747b218b79 100644 --- a/core/src/test/java/org/elasticsearch/transport/AbstractSimpleTransportTestCase.java +++ b/core/src/test/java/org/elasticsearch/transport/AbstractSimpleTransportTestCase.java @@ -37,6 +37,7 @@ import org.junit.Before; import java.io.IOException; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; @@ -204,6 +205,61 @@ public abstract class AbstractSimpleTransportTestCase extends ESTestCase { serviceA.removeHandler("sayHello"); } + public void testThreadContext() throws ExecutionException, InterruptedException { + + serviceA.registerRequestHandler("ping_pong", StringMessageRequest::new, ThreadPool.Names.GENERIC, (request, channel) -> { + assertEquals("ping_user", threadPool.getThreadContext().getHeader("test.ping.user")); + assertNull(threadPool.getThreadContext().getTransient("my_private_context")); + try { + StringMessageResponse response = new StringMessageResponse("pong"); + threadPool.getThreadContext().putHeader("test.pong.user", "pong_user"); + channel.sendResponse(response); + } catch (IOException e) { + assertThat(e.getMessage(), false, equalTo(true)); + } + }); + final Object context = new Object(); + final String executor = randomFrom(ThreadPool.THREAD_POOL_TYPES.keySet().toArray(new String[0])); + BaseTransportResponseHandler<StringMessageResponse> baseTransportResponseHandler = new BaseTransportResponseHandler<StringMessageResponse>() { + @Override + public StringMessageResponse newInstance() { + return new StringMessageResponse(); + } + + @Override + public String executor() { + return executor; + } + + @Override + public void handleResponse(StringMessageResponse response) { + assertThat("pong", equalTo(response.message)); + assertEquals("ping_user", threadPool.getThreadContext().getHeader("test.ping.user")); + assertNull(threadPool.getThreadContext().getHeader("test.pong.user")); + assertSame(context, threadPool.getThreadContext().getTransient("my_private_context")); + threadPool.getThreadContext().putHeader("some.temp.header", "booooom"); + } + + @Override + public void handleException(TransportException exp) { + assertThat("got exception instead of a response: " + exp.getMessage(), false, equalTo(true)); + } + }; + StringMessageRequest ping = new StringMessageRequest("ping"); + threadPool.getThreadContext().putHeader("test.ping.user", "ping_user"); + threadPool.getThreadContext().putTransient("my_private_context", context); + + TransportFuture<StringMessageResponse> res = serviceB.submitRequest(nodeA, "ping_pong", ping, baseTransportResponseHandler); + + StringMessageResponse message = res.get(); + assertThat("pong", equalTo(message.message)); + assertEquals("ping_user", threadPool.getThreadContext().getHeader("test.ping.user")); + assertSame(context, threadPool.getThreadContext().getTransient("my_private_context")); + assertNull("this header is only visible in the handler context", threadPool.getThreadContext().getHeader("some.temp.header")); + + serviceA.removeHandler("sayHello"); + } + public void testLocalNodeConnection() throws InterruptedException { assertTrue("serviceA is not connected to nodeA", serviceA.nodeConnected(nodeA)); if (((TransportService) serviceA).getLocalNode() != null) { diff --git a/core/src/test/java/org/elasticsearch/transport/TransportMessageTests.java b/core/src/test/java/org/elasticsearch/transport/TransportMessageTests.java deleted file mode 100644 index a94b06f6f0..0000000000 --- a/core/src/test/java/org/elasticsearch/transport/TransportMessageTests.java +++ /dev/null @@ -1,92 +0,0 @@ -/* - * Licensed to Elasticsearch under one or more contributor - * license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright - * ownership. Elasticsearch licenses this file to you under - * the Apache License, Version 2.0 (the "License"); you may - * not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.elasticsearch.transport; - -import org.elasticsearch.Version; -import org.elasticsearch.common.io.stream.BytesStreamOutput; -import org.elasticsearch.common.io.stream.StreamInput; -import org.elasticsearch.test.ESTestCase; - -import static org.hamcrest.Matchers.equalTo; -import static org.hamcrest.Matchers.is; - -/** - * - */ -public class TransportMessageTests extends ESTestCase { - public void testSerialization() throws Exception { - Message message = new Message(); - message.putHeader("key1", "value1"); - message.putHeader("key2", "value2"); - message.putInContext("key3", "value3"); - - BytesStreamOutput out = new BytesStreamOutput(); - out.setVersion(Version.CURRENT); - message.writeTo(out); - StreamInput in = StreamInput.wrap(out.bytes()); - in.setVersion(Version.CURRENT); - message = new Message(); - message.readFrom(in); - assertThat(message.getHeaders().size(), is(2)); - assertThat((String) message.getHeader("key1"), equalTo("value1")); - assertThat((String) message.getHeader("key2"), equalTo("value2")); - assertThat(message.isContextEmpty(), is(true)); - - // ensure that casting is not needed - String key1 = message.getHeader("key1"); - assertThat(key1, is("value1")); - } - - public void testCopyHeadersAndContext() throws Exception { - Message m1 = new Message(); - m1.putHeader("key1", "value1"); - m1.putHeader("key2", "value2"); - m1.putInContext("key3", "value3"); - - Message m2 = new Message(m1); - - assertThat(m2.getHeaders().size(), is(2)); - assertThat((String) m2.getHeader("key1"), equalTo("value1")); - assertThat((String) m2.getHeader("key2"), equalTo("value2")); - assertThat((String) m2.getFromContext("key3"), equalTo("value3")); - - // ensure that casting is not needed - String key3 = m2.getFromContext("key3"); - assertThat(key3, is("value3")); - testContext(m2, "key3", "value3"); - } - - // ensure that generic arg like this is not needed: TransportMessage<?> transportMessage - private void testContext(TransportMessage transportMessage, String key, String expectedValue) { - String result = transportMessage.getFromContext(key); - assertThat(result, is(expectedValue)); - - } - - private static class Message extends TransportMessage<Message> { - - private Message() { - } - - private Message(Message message) { - super(message); - } - } -} diff --git a/core/src/test/java/org/elasticsearch/transport/netty/NettyTransportIT.java b/core/src/test/java/org/elasticsearch/transport/netty/NettyTransportIT.java index 55f9bc49df..f7b8ede602 100644 --- a/core/src/test/java/org/elasticsearch/transport/netty/NettyTransportIT.java +++ b/core/src/test/java/org/elasticsearch/transport/netty/NettyTransportIT.java @@ -22,6 +22,7 @@ import org.elasticsearch.ElasticsearchException; import org.elasticsearch.Version; import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse; import org.elasticsearch.client.Client; +import org.elasticsearch.client.transport.TransportClient; import org.elasticsearch.cluster.health.ClusterHealthStatus; import org.elasticsearch.common.component.Lifecycle; import org.elasticsearch.common.inject.Inject; @@ -34,6 +35,7 @@ import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.transport.InetSocketTransportAddress; import org.elasticsearch.common.util.BigArrays; import org.elasticsearch.common.util.concurrent.AbstractRunnable; +import org.elasticsearch.common.util.concurrent.ThreadContext; import org.elasticsearch.plugins.Plugin; import org.elasticsearch.test.ESIntegTestCase; import org.elasticsearch.test.ESIntegTestCase.ClusterScope; @@ -49,6 +51,7 @@ import org.jboss.netty.channel.ChannelPipelineFactory; import java.io.IOException; import java.net.InetSocketAddress; import java.util.Collection; +import java.util.Collections; import static org.elasticsearch.common.settings.Settings.settingsBuilder; import static org.hamcrest.Matchers.containsString; @@ -78,9 +81,8 @@ public class NettyTransportIT extends ESIntegTestCase { Client transportClient = internalCluster().transportClient(); ClusterHealthResponse clusterIndexHealths = transportClient.admin().cluster().prepareHealth().get(); assertThat(clusterIndexHealths.getStatus(), is(ClusterHealthStatus.GREEN)); - try { - transportClient.admin().cluster().prepareHealth().putHeader("ERROR", "MY MESSAGE").get(); + transportClient.filterWithHeader(Collections.singletonMap("ERROR", "MY MESSAGE")).admin().cluster().prepareHealth().get(); fail("Expected exception, but didnt happen"); } catch (ElasticsearchException e) { assertThat(e.getMessage(), containsString("MY MESSAGE")); @@ -141,8 +143,9 @@ public class NettyTransportIT extends ESIntegTestCase { final TransportRequest request = reg.newRequest(); request.remoteAddress(new InetSocketTransportAddress((InetSocketAddress) channel.getRemoteAddress())); request.readFrom(buffer); - if (request.hasHeader("ERROR")) { - throw new ElasticsearchException((String) request.getHeader("ERROR")); + String error = threadPool.getThreadContext().getHeader("ERROR"); + if (error != null) { + throw new ElasticsearchException(error); } if (reg.getExecutor() == ThreadPool.Names.SAME) { //noinspection unchecked |