You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by al...@apache.org on 2021/07/30 07:45:47 UTC
[ignite] branch master updated: IGNITE-15101 Fix security context
propagation for compute tasks - Fixes #9263.
This is an automated email from the ASF dual-hosted git repository.
alexpl pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 85be1cc IGNITE-15101 Fix security context propagation for compute tasks - Fixes #9263.
85be1cc is described below
commit 85be1cc9bf14192a46cac1dbd981e4f4682648b0
Author: Mikhail Petrov <pm...@gmail.com>
AuthorDate: Fri Jul 30 10:18:17 2021 +0300
IGNITE-15101 Fix security context propagation for compute tasks - Fixes #9263.
Signed-off-by: Aleksey Plekhanov <pl...@gmail.com>
---
.../ComputeTaskRemoteSecurityContextTest.java | 527 +++++++++++++++++++++
.../internal/TaskEventSubjectIdSelfTest.java | 422 -----------------
.../client/suite/IgniteClientTestSuite.java | 6 +-
.../java/org/apache/ignite/events/TaskEvent.java | 11 +-
.../ignite/internal/GridJobExecuteRequest.java | 53 +--
.../apache/ignite/internal/GridJobSessionImpl.java | 5 -
.../ignite/internal/GridTaskSessionImpl.java | 11 -
.../ignite/internal/GridTaskSessionInternal.java | 6 -
.../apache/ignite/internal/IgniteComputeImpl.java | 24 +-
.../org/apache/ignite/internal/IgniteKernal.java | 6 +-
.../internal/cluster/ClusterGroupAdapter.java | 54 +--
.../ignite/internal/cluster/ClusterGroupEx.java | 9 -
.../ignite/internal/cluster/IgniteClusterImpl.java | 4 +-
.../managers/discovery/GridDiscoveryManager.java | 2 +
.../processors/closure/GridClosureProcessor.java | 6 +-
.../internal/processors/job/GridJobProcessor.java | 6 +-
.../internal/processors/job/GridJobWorker.java | 4 +-
.../platform/client/compute/ClientComputeTask.java | 4 -
.../client/service/ClientServiceInvokeRequest.java | 3 -
.../internal/processors/pool/PoolProcessor.java | 129 ++++-
.../processors/rest/GridRestProcessor.java | 25 +-
.../rest/handlers/task/GridTaskCommandHandler.java | 18 +-
.../handlers/user/UserActionCommandHandler.java | 2 -
.../processors/rest/request/GridRestRequest.java | 19 -
.../processors/security/IgniteSecurity.java | 7 +
.../security/IgniteSecurityProcessor.java | 89 +++-
.../security/NoOpIgniteSecurityProcessor.java | 5 +
.../processors/security/SecurityUtils.java | 8 +
.../security/thread/SecurityAwareCallable.java | 65 +++
.../security/thread/SecurityAwareIoPool.java | 62 +++
.../security/thread/SecurityAwareRunnable.java | 54 +++
.../thread/SecurityAwareStripedExecutor.java | 77 +++
.../SecurityAwareStripedThreadPoolExecutor.java | 49 ++
.../thread/SecurityAwareThreadPoolExecutor.java | 103 ++++
.../session/GridTaskSessionProcessor.java | 4 -
.../processors/task/GridTaskProcessor.java | 15 +-
.../processors/task/GridTaskThreadContextKey.java | 3 -
.../internal/processors/task/GridTaskWorker.java | 3 +-
.../ignite/internal/GridJobSubjectIdSelfTest.java | 210 --------
.../internal/managers/GridManagerStopSelfTest.java | 2 +
.../impl/TestCertificateSecurityProcessor.java | 14 +-
.../security/impl/TestSecurityProcessor.java | 15 +-
.../util/future/GridFutureAdapterSelfTest.java | 3 +
.../spi/discovery/tcp/TestReconnectProcessor.java | 5 +
.../junits/multijvm/IgniteClusterProcessProxy.java | 6 -
.../testsuites/IgniteComputeGridTestSuite.java | 2 -
46 files changed, 1269 insertions(+), 888 deletions(-)
diff --git a/modules/clients/src/test/java/org/apache/ignite/common/ComputeTaskRemoteSecurityContextTest.java b/modules/clients/src/test/java/org/apache/ignite/common/ComputeTaskRemoteSecurityContextTest.java
new file mode 100644
index 0000000..65d8ef4
--- /dev/null
+++ b/modules/clients/src/test/java/org/apache/ignite/common/ComputeTaskRemoteSecurityContextTest.java
@@ -0,0 +1,527 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.ignite.common;
+
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.LineNumberReader;
+import java.net.URL;
+import java.net.URLConnection;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.stream.Collectors;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteCompute;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.client.ClientCompute;
+import org.apache.ignite.client.Config;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.ComputeJobContext;
+import org.apache.ignite.compute.ComputeJobResult;
+import org.apache.ignite.compute.ComputeJobResultPolicy;
+import org.apache.ignite.compute.ComputeTask;
+import org.apache.ignite.compute.ComputeTaskMapAsync;
+import org.apache.ignite.compute.ComputeTaskTimeoutException;
+import org.apache.ignite.configuration.ClientConfiguration;
+import org.apache.ignite.configuration.ClientConnectorConfiguration;
+import org.apache.ignite.configuration.ConnectorConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.ThinClientConfiguration;
+import org.apache.ignite.events.Event;
+import org.apache.ignite.events.JobEvent;
+import org.apache.ignite.events.TaskEvent;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.client.GridClient;
+import org.apache.ignite.internal.client.GridClientCompute;
+import org.apache.ignite.internal.client.GridClientConfiguration;
+import org.apache.ignite.internal.client.GridClientFactory;
+import org.apache.ignite.internal.client.GridClientNode;
+import org.apache.ignite.internal.client.thin.ClientServerError;
+import org.apache.ignite.internal.processors.rest.GridRestCommand;
+import org.apache.ignite.internal.processors.rest.protocols.http.jetty.GridJettyObjectMapper;
+import org.apache.ignite.internal.processors.security.AbstractSecurityTest;
+import org.apache.ignite.internal.processors.security.AbstractTestSecurityPluginProvider;
+import org.apache.ignite.internal.util.typedef.X;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.plugin.security.SecurityCredentials;
+import org.apache.ignite.plugin.security.SecurityCredentialsBasicProvider;
+import org.apache.ignite.plugin.security.SecurityPermissionSet;
+import org.apache.ignite.resources.IgniteInstanceResource;
+import org.apache.ignite.resources.JobContextResource;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import org.junit.Assume;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.Collections.singletonList;
+import static org.apache.ignite.Ignition.allGrids;
+import static org.apache.ignite.events.EventType.EVT_JOB_CANCELLED;
+import static org.apache.ignite.events.EventType.EVT_JOB_FAILED;
+import static org.apache.ignite.events.EventType.EVT_JOB_FINISHED;
+import static org.apache.ignite.events.EventType.EVT_JOB_MAPPED;
+import static org.apache.ignite.events.EventType.EVT_JOB_QUEUED;
+import static org.apache.ignite.events.EventType.EVT_JOB_RESULTED;
+import static org.apache.ignite.events.EventType.EVT_JOB_STARTED;
+import static org.apache.ignite.events.EventType.EVT_TASK_FAILED;
+import static org.apache.ignite.events.EventType.EVT_TASK_FINISHED;
+import static org.apache.ignite.events.EventType.EVT_TASK_REDUCED;
+import static org.apache.ignite.events.EventType.EVT_TASK_STARTED;
+import static org.apache.ignite.events.EventType.EVT_TASK_TIMEDOUT;
+import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_IGNITE_INSTANCE_NAME;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+
+/** Tests that compute tasks are executed with the security context of the task initiator. */
+@RunWith(Parameterized.class)
+public class ComputeTaskRemoteSecurityContextTest extends AbstractSecurityTest {
+ /** Port for REST client connection. */
+ private static final String DFLT_REST_PORT = "11080";
+
+ /** Task timeout.*/
+ private static final long TEST_TASK_TIMEOUT = 500;
+
+ /** Custom object mapper for HTTP REST API. */
+ private static final ObjectMapper OBJECT_MAPPER = new GridJettyObjectMapper();
+
+ /** Events paired with the nodes on which they were listened to. */
+ private static final Map<ClusterNode, Collection<Event>> LISTENED_TASK_EVENTS = new HashMap<>();
+
+ /** Indicates whether the test task can proceed with its execution. */
+ private static CountDownLatch taskExecutionUnlockedLatch;
+
+ /** All possible task events. */
+ private static final int[] TASK_EVENTS = {
+ EVT_TASK_STARTED, EVT_TASK_FINISHED, EVT_TASK_REDUCED, EVT_JOB_MAPPED, EVT_JOB_RESULTED, EVT_JOB_STARTED,
+ EVT_JOB_FINISHED, EVT_JOB_QUEUED, EVT_TASK_TIMEDOUT, EVT_TASK_FAILED, EVT_JOB_CANCELLED, EVT_JOB_FAILED};
+
+ /** Events that occur on the task reducer if task execution completes successfully. */
+ private static final List<Integer> REDUCER_SUCCEEDED_TASK_EVENTS = Arrays.asList(
+ EVT_JOB_QUEUED, EVT_JOB_STARTED, EVT_JOB_FINISHED, EVT_TASK_STARTED, EVT_JOB_MAPPED, EVT_TASK_REDUCED,
+ EVT_JOB_RESULTED, EVT_TASK_FINISHED);
+
+ /** Events that occur on task map node if task execution fails. */
+ private static final List<Integer> REDUCER_FAILED_TASK_EVENTS = Arrays.asList(
+ EVT_JOB_QUEUED, EVT_JOB_STARTED, EVT_JOB_CANCELLED, EVT_JOB_FAILED, EVT_TASK_STARTED, EVT_JOB_MAPPED,
+ EVT_TASK_TIMEDOUT, EVT_TASK_FAILED);
+
+ /** Events that occur on task reducer if task execution completes successfully. */
+ private static final List<Integer> MAP_NODE_SUCCEEDED_TASK_EVENTS = Arrays.asList(
+ EVT_JOB_QUEUED, EVT_JOB_STARTED, EVT_JOB_FINISHED);
+
+ /** Events that occur on task map node if task execution fails. */
+ private static final List<Integer> MAP_NODE_FAILED_TASK_EVENTS = Arrays.asList(
+ EVT_JOB_QUEUED, EVT_JOB_STARTED, EVT_JOB_CANCELLED, EVT_JOB_FAILED);
+
+ /** */
+ @Parameterized.Parameters(name = "async={0} failWithTimeout={1} mapAsync={2}")
+ public static Iterable<Boolean[]> data() {
+ List<Boolean[]> res = new ArrayList<>();
+
+ for (Boolean async : Arrays.asList(false, true)) {
+ for (Boolean failWithTimeout : Arrays.asList(false, true)) {
+ for (Boolean mapAsync : Arrays.asList(false, true))
+ res.add(new Boolean[] {async, failWithTimeout, mapAsync});
+ }
+ }
+
+ return res;
+ }
+
+ /** Whether task is executed asynchronously. */
+ @Parameterized.Parameter()
+ public boolean async;
+
+ /** Whether task fails with timeout exception. */
+ @Parameterized.Parameter(1)
+ public boolean failWithTimeout;
+
+ /** Whether task mapping is processed asynchronously. */
+ @Parameterized.Parameter(2)
+ public boolean mapAsync;
+
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ LISTENED_TASK_EVENTS.put(startGridAllowAll("crd").localNode(), ConcurrentHashMap.newKeySet());
+ LISTENED_TASK_EVENTS.put(startGridAllowAll("srv").localNode(), ConcurrentHashMap.newKeySet());
+ LISTENED_TASK_EVENTS.put(startGridAllowAll("cli").localNode(), ConcurrentHashMap.newKeySet());
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ if (taskExecutionUnlockedLatch != null)
+ taskExecutionUnlockedLatch.countDown();
+
+ super.afterTest();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ LISTENED_TASK_EVENTS.values().forEach(Collection::clear);
+
+ taskExecutionUnlockedLatch = null;
+
+ if (failWithTimeout)
+ taskExecutionUnlockedLatch = new CountDownLatch(1);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(
+ String instanceName,
+ AbstractTestSecurityPluginProvider pluginProv
+ ) throws Exception {
+ return super.getConfiguration(instanceName, pluginProv)
+ .setLocalHost("127.0.0.1")
+ .setIncludeEventTypes(TASK_EVENTS)
+ .setConnectorConfiguration(new ConnectorConfiguration()
+ .setJettyPath("modules/clients/src/test/resources/jetty/rest-jetty.xml"))
+ .setClientConnectorConfiguration(new ClientConnectorConfiguration()
+ .setThinClientConfiguration(new ThinClientConfiguration()
+ .setMaxActiveComputeTasksPerConnection(1)));
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteEx startGrid(String login, SecurityPermissionSet prmSet, boolean isClient) throws Exception {
+ IgniteEx ignite = startGrid(login, prmSet, null, isClient);
+
+ ignite.events().localListen(evt -> {
+ LISTENED_TASK_EVENTS.get(ignite.localNode()).add(evt);
+
+ return true;
+ }, TASK_EVENTS);
+
+ return ignite;
+ }
+
+ /** Tests task execution security context in case task was initiated from the {@link GridClient}. */
+ @Test
+ public void testGridClient() throws Exception {
+ Assume.assumeFalse(failWithTimeout);
+
+ String login = "grid_client";
+
+ GridClientConfiguration cfg = new GridClientConfiguration()
+ .setServers(singletonList("127.0.0.1:11211"))
+ .setSecurityCredentialsProvider(new SecurityCredentialsBasicProvider(new SecurityCredentials(login, "")));
+
+ try (GridClient cli = GridClientFactory.start(cfg)) {
+ GridClientNode taskReqRecipient = cli.compute().nodes().stream()
+ .filter(n -> "crd".equals(n.attribute(ATTR_IGNITE_INSTANCE_NAME)))
+ .findFirst().orElseThrow(NoSuchElementException::new);
+
+ GridClientCompute comp = cli.compute().projection(taskReqRecipient);
+
+ String taskName = mapAsync ? MapAsyncTestTask.class.getName() : TestTask.class.getName();
+
+ if (async)
+ comp.executeAsync(taskName, login).get();
+ else
+ comp.execute(taskName, login);
+
+ checkTaskEvents("crd", login, REDUCER_SUCCEEDED_TASK_EVENTS, MAP_NODE_SUCCEEDED_TASK_EVENTS);
+ }
+ }
+
+ /** Tests task execution security context in case task was initiated from the {@link IgniteClient}. */
+ @Test
+ public void testIgniteClient() throws Exception {
+ String login = "thin_client";
+
+ ClientConfiguration cfg = new ClientConfiguration()
+ .setAddresses(Config.SERVER)
+ .setUserName(login)
+ .setUserPassword("");
+
+ try (IgniteClient cli = Ignition.startClient(cfg)) {
+ ClientCompute comp = cli.compute(cli.cluster().forNodes(cli.cluster().nodes()));
+
+ if (failWithTimeout)
+ comp = comp.withTimeout(TEST_TASK_TIMEOUT);
+
+ String taskName = mapAsync ? MapAsyncTestTask.class.getName() : TestTask.class.getName();
+
+ Throwable timeoutE = null;
+
+ try {
+ if (async)
+ comp.executeAsync2(taskName, login).get();
+ else
+ comp.execute(taskName, login);
+
+ checkTaskEvents("crd", login, REDUCER_SUCCEEDED_TASK_EVENTS, MAP_NODE_SUCCEEDED_TASK_EVENTS);
+ }
+ catch (Throwable e) {
+ if (!failWithTimeout)
+ throw e;
+
+ timeoutE = e;
+ }
+
+ if (failWithTimeout) {
+ assertNotNull(timeoutE);
+ assertTrue(X.hasCause(timeoutE, "Task timed out", ClientServerError.class));
+
+ checkTaskEvents("crd", login, REDUCER_FAILED_TASK_EVENTS, MAP_NODE_FAILED_TASK_EVENTS);
+ }
+ }
+ }
+
+ /** Tests task execution security context in case task was initiated from the REST client. */
+ @Test
+ public void testRestClient() throws Exception {
+ String login = "rest_client";
+
+ String taskName = mapAsync ? MapAsyncTestTask.class.getName() : TestTask.class.getName();
+
+ JsonNode resp = sendRestRequest("http://127.0.0.1:" + DFLT_REST_PORT + "/ignite" +
+ "?ignite.login=" + login +
+ "&ignite.password=" +
+ "&cmd=" + GridRestCommand.EXE.key() +
+ "&name=" + taskName +
+ "&p1=" + login +
+ "&async=" + async +
+ "&timeout=" + (failWithTimeout ? TEST_TASK_TIMEOUT : 0));
+
+ if (async) {
+ String taskId = resp.get("response").get("id").textValue();
+
+ U.sleep(2 * TEST_TASK_TIMEOUT);
+
+ resp = sendRestRequest("http://127.0.0.1:" + DFLT_REST_PORT + "/ignite" +
+ "?ignite.login=" + login +
+ "&ignite.password=" +
+ "&id=" + taskId +
+ "&cmd=" + GridRestCommand.RESULT.key());
+ }
+
+ if (failWithTimeout) {
+ assertEquals(1, resp.get("successStatus").intValue());
+ assertTrue(resp.get("error").textValue().contains("Task timed out"));
+
+ checkTaskEvents("crd", login, REDUCER_FAILED_TASK_EVENTS, MAP_NODE_FAILED_TASK_EVENTS);
+ }
+ else {
+ assertEquals(0, resp.get("successStatus").intValue());
+ assertNull(resp.get("error").textValue());
+
+ JsonNode taskRes = resp.get("response");
+
+ assertTrue(taskRes.get("finished").asBoolean());
+ assertNull(taskRes.get("error").textValue());
+
+ checkTaskEvents("crd", login, REDUCER_SUCCEEDED_TASK_EVENTS, MAP_NODE_SUCCEEDED_TASK_EVENTS);
+ }
+ }
+
+ /** Tests task execution security context in case task was initiated from the {@link Ignite} server node. */
+ @Test
+ public void testServerNode() throws Exception {
+ doNodeTest(false);
+ }
+
+
+ /** Tests task execution security context in case task was initiated from the {@link Ignite} client node. */
+ @Test
+ public void testClientNode() throws Exception {
+ doNodeTest(true);
+ }
+
+ /** */
+ public void doNodeTest(boolean isClient) throws Exception {
+ String login = isClient ? "cli" : "srv";
+
+ IgniteEx ignite = grid(login);
+
+ IgniteCompute compute = ignite.compute(ignite.cluster().forNodes(ignite.cluster().nodes()));
+
+ if (failWithTimeout)
+ compute.withTimeout(TEST_TASK_TIMEOUT);
+
+ String taskName = mapAsync ? MapAsyncTestTask.class.getName() : TestTask.class.getName();
+
+ ComputeTaskTimeoutException timeoutE = null;
+
+ try {
+ if (async)
+ compute.executeAsync(taskName, login).get();
+ else
+ compute.execute(taskName, login);
+
+ checkTaskEvents(login, login, REDUCER_SUCCEEDED_TASK_EVENTS, MAP_NODE_SUCCEEDED_TASK_EVENTS);
+ }
+ catch (ComputeTaskTimeoutException e) {
+ if (!failWithTimeout)
+ throw e;
+
+ timeoutE = e;
+ }
+
+ if (failWithTimeout) {
+ assertNotNull(timeoutE);
+
+ checkTaskEvents(login, login, REDUCER_FAILED_TASK_EVENTS, MAP_NODE_FAILED_TASK_EVENTS);
+ }
+ }
+
+ /** */
+ private void checkTaskEvents(
+ String taskReducerNode,
+ String taskInitiatorLogin,
+ List<Integer> expReducerTaskEvts,
+ List<Integer> expMapNodeTaskEvts
+ ) throws Exception {
+ for (Ignite ignite : allGrids()) {
+ checkEvents(((IgniteEx)ignite).localNode(), taskInitiatorLogin, ignite.name().equals(taskReducerNode)
+ ? expReducerTaskEvts : expMapNodeTaskEvts);
+ }
+ }
+
+ /** */
+ private void checkEvents(ClusterNode node, String expLogin, List<Integer> expEvtTypes) throws Exception {
+ Collection<Event> nodeTaskEvents = LISTENED_TASK_EVENTS.get(node);
+
+ assertTrue(waitForCondition(() ->
+ nodeTaskEvents.stream()
+ .map(Event::type)
+ .collect(Collectors.toList())
+ .containsAll(expEvtTypes), getTestTimeout()));
+
+ assertTrue(nodeTaskEvents.stream()
+ .map(evt -> evt instanceof TaskEvent ? ((TaskEvent)evt).subjectId() : ((JobEvent)evt).taskSubjectId())
+ .map(subjId -> {
+ try {
+ return grid("crd").context().security().authenticatedSubject(subjId).login();
+ }
+ catch (IgniteCheckedException e) {
+ throw new SecurityException(e);
+ }
+ })
+ .allMatch(expLogin::equals));
+ }
+
+ /** */
+ static JsonNode sendRestRequest(String url) throws IOException {
+ URLConnection conn = new URL(url).openConnection();
+
+ StringBuilder buf = new StringBuilder(256);
+
+ try (LineNumberReader rdr = new LineNumberReader(new InputStreamReader(conn.getInputStream(), UTF_8))) {
+ for (String line = rdr.readLine(); line != null; line = rdr.readLine())
+ buf.append(line);
+ }
+
+ return OBJECT_MAPPER.readTree(buf.toString());
+ }
+
+ /** */
+ @ComputeTaskMapAsync
+ public static class MapAsyncTestTask extends TestTask {
+ // No-op.
+ }
+
+ /** Test compute task that checks security context of the task initiator. */
+ public static class TestTask implements ComputeTask<String, Void> {
+ /** {@inheritDoc} */
+ @Override public @NotNull Map<? extends ComputeJob, ClusterNode> map(
+ List<ClusterNode> subgrid,
+ @Nullable String taskInitiatorLogin
+ ) throws IgniteException {
+ Map<ComputeJob, ClusterNode> res = new HashMap<>();
+
+ for (ClusterNode node : subgrid) {
+ res.put(new ComputeJob() {
+ /** */
+ @IgniteInstanceResource
+ private IgniteEx ignite;
+
+ /** */
+ @JobContextResource
+ private ComputeJobContext jobCtx;
+
+ /** */
+ private boolean suspended;
+
+ @Override public void cancel() {
+ // No-op.
+ }
+
+ @Override public Object execute() throws IgniteException {
+ assertEquals(taskInitiatorLogin, ignite.context().security().securityContext().subject().login());
+
+ if (taskExecutionUnlockedLatch != null) {
+ try {
+ taskExecutionUnlockedLatch.await();
+ }
+ catch (InterruptedException e) {
+ throw new IgniteException();
+ }
+
+ return null;
+ }
+
+ if (!suspended) {
+ jobCtx.holdcc();
+
+ suspended = true;
+
+ new Thread(() -> jobCtx.callcc()).start();
+
+ return null;
+ }
+
+ return null;
+ }
+ }, node);
+ }
+
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override public ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult> rcvd) {
+ if (res.getException() != null)
+ throw new IgniteException(res.getException());
+
+ return ComputeJobResultPolicy.WAIT;
+ }
+
+ /** {@inheritDoc} */
+ @Override public @Nullable Void reduce(List<ComputeJobResult> results) throws IgniteException {
+ assertEquals(allGrids().size(), results.size());
+
+ return null;
+ }
+ }
+}
diff --git a/modules/clients/src/test/java/org/apache/ignite/internal/TaskEventSubjectIdSelfTest.java b/modules/clients/src/test/java/org/apache/ignite/internal/TaskEventSubjectIdSelfTest.java
deleted file mode 100644
index 03c4916..0000000
--- a/modules/clients/src/test/java/org/apache/ignite/internal/TaskEventSubjectIdSelfTest.java
+++ /dev/null
@@ -1,422 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF 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.apache.ignite.internal;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
-import java.util.UUID;
-import java.util.concurrent.Callable;
-import java.util.concurrent.CountDownLatch;
-import org.apache.ignite.Ignite;
-import org.apache.ignite.IgniteException;
-import org.apache.ignite.compute.ComputeJob;
-import org.apache.ignite.compute.ComputeJobAdapter;
-import org.apache.ignite.compute.ComputeJobResult;
-import org.apache.ignite.compute.ComputeTaskSplitAdapter;
-import org.apache.ignite.compute.ComputeTaskTimeoutException;
-import org.apache.ignite.configuration.ConnectorConfiguration;
-import org.apache.ignite.configuration.IgniteConfiguration;
-import org.apache.ignite.events.Event;
-import org.apache.ignite.events.EventType;
-import org.apache.ignite.events.TaskEvent;
-import org.apache.ignite.internal.client.GridClient;
-import org.apache.ignite.internal.client.GridClientConfiguration;
-import org.apache.ignite.internal.client.GridClientFactory;
-import org.apache.ignite.lang.IgnitePredicate;
-import org.apache.ignite.lang.IgniteRunnable;
-import org.apache.ignite.testframework.GridTestUtils;
-import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
-import org.jetbrains.annotations.Nullable;
-import org.junit.Test;
-
-import static java.util.concurrent.TimeUnit.MILLISECONDS;
-import static org.apache.ignite.events.EventType.EVTS_TASK_EXECUTION;
-import static org.apache.ignite.events.EventType.EVT_TASK_FAILED;
-import static org.apache.ignite.events.EventType.EVT_TASK_FINISHED;
-import static org.apache.ignite.events.EventType.EVT_TASK_REDUCED;
-import static org.apache.ignite.events.EventType.EVT_TASK_STARTED;
-import static org.apache.ignite.events.EventType.EVT_TASK_TIMEDOUT;
-
-/**
- * Tests for security subject ID in task events.
- */
-public class TaskEventSubjectIdSelfTest extends GridCommonAbstractTest {
- /** */
- private static final Collection<TaskEvent> evts = new ArrayList<>();
-
- /** */
- private static CountDownLatch latch;
-
- /** */
- private static UUID nodeId;
-
- /** */
- private static GridClient client;
-
- /** {@inheritDoc} */
- @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
- IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
-
- cfg.setConnectorConfiguration(new ConnectorConfiguration());
-
- cfg.setIncludeEventTypes(EventType.EVTS_ALL);
-
- return cfg;
- }
-
- /** {@inheritDoc} */
- @Override protected void beforeTestsStarted() throws Exception {
- Ignite g = startGrid();
-
- g.events().localListen(new IgnitePredicate<Event>() {
- @Override public boolean apply(Event evt) {
- assert evt instanceof TaskEvent;
-
- evts.add((TaskEvent)evt);
-
- latch.countDown();
-
- return true;
- }
- }, EVTS_TASK_EXECUTION);
-
- nodeId = g.cluster().localNode().id();
-
- GridClientConfiguration cfg = new GridClientConfiguration();
-
- cfg.setServers(Collections.singleton("127.0.0.1:11211"));
-
- client = GridClientFactory.start(cfg);
- }
-
- /** {@inheritDoc} */
- @Override protected void afterTestsStopped() throws Exception {
- GridClientFactory.stop(client.id());
- }
-
- /** {@inheritDoc} */
- @Override protected void beforeTest() throws Exception {
- evts.clear();
- }
-
- /**
- * @throws Exception If failed.
- */
- @Test
- public void testSimpleTask() throws Exception {
- latch = new CountDownLatch(3);
-
- grid().compute().execute(new SimpleTask(), null);
-
- assert latch.await(1000, MILLISECONDS);
-
- assertEquals(3, evts.size());
-
- Iterator<TaskEvent> it = evts.iterator();
-
- assert it.hasNext();
-
- TaskEvent evt = it.next();
-
- assert evt != null;
-
- assertEquals(EVT_TASK_STARTED, evt.type());
- assertEquals(nodeId, evt.subjectId());
-
- assert it.hasNext();
-
- evt = it.next();
-
- assert evt != null;
-
- assertEquals(EVT_TASK_REDUCED, evt.type());
- assertEquals(nodeId, evt.subjectId());
-
- assert it.hasNext();
-
- evt = it.next();
-
- assert evt != null;
-
- assertEquals(EVT_TASK_FINISHED, evt.type());
- assertEquals(nodeId, evt.subjectId());
-
- assert !it.hasNext();
- }
-
- /**
- * @throws Exception If failed.
- */
- @Test
- public void testFailedTask() throws Exception {
- latch = new CountDownLatch(2);
-
- GridTestUtils.assertThrows(
- log,
- new Callable<Object>() {
- @Override public Object call() throws Exception {
- grid().compute().execute(new FailedTask(), null);
-
- return null;
- }
- },
- IgniteException.class,
- null
- );
-
- assert latch.await(1000, MILLISECONDS);
-
- assertEquals(2, evts.size());
-
- Iterator<TaskEvent> it = evts.iterator();
-
- assert it.hasNext();
-
- TaskEvent evt = it.next();
-
- assert evt != null;
-
- assertEquals(EVT_TASK_STARTED, evt.type());
- assertEquals(nodeId, evt.subjectId());
-
- assert it.hasNext();
-
- evt = it.next();
-
- assert evt != null;
-
- assertEquals(EVT_TASK_FAILED, evt.type());
- assertEquals(nodeId, evt.subjectId());
-
- assert !it.hasNext();
- }
-
- /**
- * @throws Exception If failed.
- */
- @Test
- public void testTimedOutTask() throws Exception {
- latch = new CountDownLatch(2);
-
- GridTestUtils.assertThrows(
- log,
- new Callable<Object>() {
- @Override public Object call() throws Exception {
- grid().compute().withTimeout(100).execute(new TimedOutTask(), null);
-
- return null;
- }
- },
- ComputeTaskTimeoutException.class,
- null
- );
-
- assert latch.await(1000, MILLISECONDS);
-
- assertEquals(3, evts.size());
-
- Iterator<TaskEvent> it = evts.iterator();
-
- assert it.hasNext();
-
- TaskEvent evt = it.next();
-
- assert evt != null;
-
- assertEquals(EVT_TASK_STARTED, evt.type());
- assertEquals(nodeId, evt.subjectId());
-
- assert it.hasNext();
-
- evt = it.next();
-
- assert evt != null;
-
- assertEquals(EVT_TASK_TIMEDOUT, evt.type());
- assertEquals(nodeId, evt.subjectId());
-
- assert it.hasNext();
-
- evt = it.next();
-
- assert evt != null;
-
- assertEquals(EVT_TASK_FAILED, evt.type());
- assertEquals(nodeId, evt.subjectId());
-
- assert !it.hasNext();
- }
-
- /**
- * @throws Exception If failed.
- */
- @Test
- public void testClosure() throws Exception {
- latch = new CountDownLatch(3);
-
- grid().compute().run(new IgniteRunnable() {
- @Override public void run() {
- // No-op.
- }
- });
-
- assert latch.await(1000, MILLISECONDS);
-
- assertEquals(3, evts.size());
-
- Iterator<TaskEvent> it = evts.iterator();
-
- assert it.hasNext();
-
- TaskEvent evt = it.next();
-
- assert evt != null;
-
- assertEquals(EVT_TASK_STARTED, evt.type());
- assertEquals(nodeId, evt.subjectId());
-
- assert it.hasNext();
-
- evt = it.next();
-
- assert evt != null;
-
- assertEquals(EVT_TASK_REDUCED, evt.type());
- assertEquals(nodeId, evt.subjectId());
-
- assert it.hasNext();
-
- evt = it.next();
-
- assert evt != null;
-
- assertEquals(EVT_TASK_FINISHED, evt.type());
- assertEquals(nodeId, evt.subjectId());
-
- assert !it.hasNext();
- }
-
- /**
- * Events for class tasks that was started from external clients should contain
- * client subject id instead of the node where it was started. This test checks it.
- *
- * @throws Exception If failed.
- */
- @Test
- public void testClient() throws Exception {
- latch = new CountDownLatch(3);
-
- client.compute().execute(SimpleTask.class.getName(), null);
-
- assert latch.await(1000, MILLISECONDS);
-
- assertEquals(3, evts.size());
-
- Iterator<TaskEvent> it = evts.iterator();
-
- assert it.hasNext();
-
- TaskEvent evt = it.next();
-
- assert evt != null;
-
- assertEquals(EVT_TASK_STARTED, evt.type());
- assertEquals(client.id(), evt.subjectId());
-
- assert it.hasNext();
-
- evt = it.next();
-
- assert evt != null;
-
- assertEquals(EVT_TASK_REDUCED, evt.type());
- assertEquals(client.id(), evt.subjectId());
-
- assert it.hasNext();
-
- evt = it.next();
-
- assert evt != null;
-
- assertEquals(EVT_TASK_FINISHED, evt.type());
- assertEquals(client.id(), evt.subjectId());
-
- assert !it.hasNext();
- }
-
- /** */
- private static class SimpleTask extends ComputeTaskSplitAdapter<Object, Object> {
- /** {@inheritDoc} */
- @Override protected Collection<? extends ComputeJob> split(int gridSize, Object arg) {
- return Collections.singleton(new ComputeJobAdapter() {
- @Nullable @Override public Object execute() {
- return null;
- }
- });
- }
-
- /** {@inheritDoc} */
- @Nullable @Override public Object reduce(List<ComputeJobResult> results) {
- return null;
- }
- }
-
- /** */
- private static class FailedTask extends ComputeTaskSplitAdapter<Object, Object> {
- /** {@inheritDoc} */
- @Override protected Collection<? extends ComputeJob> split(int gridSize, Object arg) {
- return Collections.singleton(new ComputeJobAdapter() {
- @Nullable @Override public Object execute() {
- return null;
- }
- });
- }
-
- /** {@inheritDoc} */
- @Nullable @Override public Object reduce(List<ComputeJobResult> results) {
- throw new IgniteException("Task failed.");
- }
- }
-
- /** */
- private static class TimedOutTask extends ComputeTaskSplitAdapter<Object, Object> {
- /** {@inheritDoc} */
- @Override protected Collection<? extends ComputeJob> split(int gridSize, Object arg) {
- return Collections.singleton(new ComputeJobAdapter() {
- @Nullable @Override public Object execute() {
- try {
- Thread.sleep(10000);
- }
- catch (InterruptedException ignored) {
- // No-op.
- }
-
- return null;
- }
- });
- }
-
- /** {@inheritDoc} */
- @Nullable @Override public Object reduce(List<ComputeJobResult> results) {
- return null;
- }
- }
-}
diff --git a/modules/clients/src/test/java/org/apache/ignite/internal/client/suite/IgniteClientTestSuite.java b/modules/clients/src/test/java/org/apache/ignite/internal/client/suite/IgniteClientTestSuite.java
index 0a50e40..341271d 100644
--- a/modules/clients/src/test/java/org/apache/ignite/internal/client/suite/IgniteClientTestSuite.java
+++ b/modules/clients/src/test/java/org/apache/ignite/internal/client/suite/IgniteClientTestSuite.java
@@ -19,9 +19,9 @@ package org.apache.ignite.internal.client.suite;
import org.apache.ignite.common.ClientSideCacheCreationDestructionWileTopologyChangeTest;
import org.apache.ignite.common.ClientSizeCacheCreationDestructionTest;
+import org.apache.ignite.common.ComputeTaskRemoteSecurityContextTest;
import org.apache.ignite.common.NodeSslConnectionMetricTest;
import org.apache.ignite.internal.IgniteClientFailuresTest;
-import org.apache.ignite.internal.TaskEventSubjectIdSelfTest;
import org.apache.ignite.internal.client.ClientDefaultCacheSelfTest;
import org.apache.ignite.internal.client.ClientReconnectionSelfTest;
import org.apache.ignite.internal.client.ClientSslParametersTest;
@@ -156,7 +156,6 @@ import org.junit.runners.Suite;
// Rest task command handler test.
TaskCommandHandlerSelfTest.class,
ChangeStateCommandHandlerTest.class,
- TaskEventSubjectIdSelfTest.class,
// Default cache only test.
ClientDefaultCacheSelfTest.class,
@@ -184,7 +183,8 @@ import org.junit.runners.Suite;
ClientSizeCacheCreationDestructionTest.class,
ClientSideCacheCreationDestructionWileTopologyChangeTest.class,
- NodeSslConnectionMetricTest.class
+ NodeSslConnectionMetricTest.class,
+ ComputeTaskRemoteSecurityContextTest.class
})
public class IgniteClientTestSuite {
}
diff --git a/modules/core/src/main/java/org/apache/ignite/events/TaskEvent.java b/modules/core/src/main/java/org/apache/ignite/events/TaskEvent.java
index ef958ed..95f7fcb 100644
--- a/modules/core/src/main/java/org/apache/ignite/events/TaskEvent.java
+++ b/modules/core/src/main/java/org/apache/ignite/events/TaskEvent.java
@@ -19,6 +19,7 @@ package org.apache.ignite.events;
import java.util.UUID;
import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.processors.security.IgniteSecurity;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteUuid;
@@ -146,13 +147,11 @@ public class TaskEvent extends EventAdapter {
}
/**
- * Gets security subject ID initiated this task event, if available. This property
- * is not available for GridEventType#EVT_TASK_SESSION_ATTR_SET task event.
- * <p>
- * Subject ID will be set either to node ID or client ID initiated
- * task execution.
+ * Gets security subject ID initiated this task event, if security is enabled. This property is not available for
+ * {@link EventType#EVT_TASK_SESSION_ATTR_SET} task event.
*
- * @return Subject ID.
+ * @return Subject ID if security is enabled, otherwise null.
+ * @see IgniteSecurity#enabled()
*/
@Nullable public UUID subjectId() {
return subjId;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridJobExecuteRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/GridJobExecuteRequest.java
index 2507ac4..da32735 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridJobExecuteRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridJobExecuteRequest.java
@@ -48,9 +48,6 @@ public class GridJobExecuteRequest implements ExecutorAwareMessage {
/** */
private static final long serialVersionUID = 0L;
- /** Subject ID. */
- private UUID subjId;
-
/** */
private IgniteUuid sesId;
@@ -193,7 +190,6 @@ public class GridJobExecuteRequest implements ExecutorAwareMessage {
* @param forceLocDep {@code True} If remote node should ignore deployment settings.
* @param sesFullSup {@code True} if session attributes are disabled.
* @param internal {@code True} if internal job.
- * @param subjId Subject ID.
* @param cacheIds Caches' identifiers to reserve partition.
* @param part Partition to lock.
* @param topVer Affinity topology version of job mapping.
@@ -226,7 +222,6 @@ public class GridJobExecuteRequest implements ExecutorAwareMessage {
boolean forceLocDep,
boolean sesFullSup,
boolean internal,
- UUID subjId,
@Nullable int[] cacheIds,
int part,
@Nullable AffinityTopologyVersion topVer,
@@ -269,7 +264,6 @@ public class GridJobExecuteRequest implements ExecutorAwareMessage {
this.forceLocDep = forceLocDep;
this.sesFullSup = sesFullSup;
this.internal = internal;
- this.subjId = subjId;
this.idsOfCaches = cacheIds;
this.part = part;
this.topVer = topVer;
@@ -472,13 +466,6 @@ public class GridJobExecuteRequest implements ExecutorAwareMessage {
}
/**
- * @return Subject ID.
- */
- public UUID getSubjectId() {
- return subjId;
- }
-
- /**
* @return Caches' identifiers to reserve specified partition for job execution.
*/
public int[] getCacheIds() {
@@ -630,48 +617,42 @@ public class GridJobExecuteRequest implements ExecutorAwareMessage {
writer.incrementState();
case 18:
- if (!writer.writeUuid("subjId", subjId))
- return false;
-
- writer.incrementState();
-
- case 19:
if (!writer.writeString("taskClsName", taskClsName))
return false;
writer.incrementState();
- case 20:
+ case 19:
if (!writer.writeString("taskName", taskName))
return false;
writer.incrementState();
- case 21:
+ case 20:
if (!writer.writeLong("timeout", timeout))
return false;
writer.incrementState();
- case 22:
+ case 21:
if (!writer.writeCollection("top", top, MessageCollectionItemType.UUID))
return false;
writer.incrementState();
- case 23:
+ case 22:
if (!writer.writeByteArray("topPredBytes", topPredBytes))
return false;
writer.incrementState();
- case 24:
+ case 23:
if (!writer.writeAffinityTopologyVersion("topVer", topVer))
return false;
writer.incrementState();
- case 25:
+ case 24:
if (!writer.writeString("userVer", userVer))
return false;
@@ -839,14 +820,6 @@ public class GridJobExecuteRequest implements ExecutorAwareMessage {
reader.incrementState();
case 18:
- subjId = reader.readUuid("subjId");
-
- if (!reader.isLastRead())
- return false;
-
- reader.incrementState();
-
- case 19:
taskClsName = reader.readString("taskClsName");
if (!reader.isLastRead())
@@ -854,7 +827,7 @@ public class GridJobExecuteRequest implements ExecutorAwareMessage {
reader.incrementState();
- case 20:
+ case 19:
taskName = reader.readString("taskName");
if (!reader.isLastRead())
@@ -862,7 +835,7 @@ public class GridJobExecuteRequest implements ExecutorAwareMessage {
reader.incrementState();
- case 21:
+ case 20:
timeout = reader.readLong("timeout");
if (!reader.isLastRead())
@@ -870,7 +843,7 @@ public class GridJobExecuteRequest implements ExecutorAwareMessage {
reader.incrementState();
- case 22:
+ case 21:
top = reader.readCollection("top", MessageCollectionItemType.UUID);
if (!reader.isLastRead())
@@ -878,7 +851,7 @@ public class GridJobExecuteRequest implements ExecutorAwareMessage {
reader.incrementState();
- case 23:
+ case 22:
topPredBytes = reader.readByteArray("topPredBytes");
if (!reader.isLastRead())
@@ -886,7 +859,7 @@ public class GridJobExecuteRequest implements ExecutorAwareMessage {
reader.incrementState();
- case 24:
+ case 23:
topVer = reader.readAffinityTopologyVersion("topVer");
if (!reader.isLastRead())
@@ -894,7 +867,7 @@ public class GridJobExecuteRequest implements ExecutorAwareMessage {
reader.incrementState();
- case 25:
+ case 24:
userVer = reader.readString("userVer");
if (!reader.isLastRead())
@@ -914,7 +887,7 @@ public class GridJobExecuteRequest implements ExecutorAwareMessage {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 26;
+ return 25;
}
/** {@inheritDoc} */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridJobSessionImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/GridJobSessionImpl.java
index 6f1fb32..5d98789 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridJobSessionImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridJobSessionImpl.java
@@ -299,11 +299,6 @@ public class GridJobSessionImpl implements GridTaskSessionInternal {
}
/** {@inheritDoc} */
- @Override public UUID subjectId() {
- return ses.subjectId();
- }
-
- /** {@inheritDoc} */
@Override public IgniteFuture<?> mapFuture() {
return new IgniteFinishedFutureImpl<>(null);
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridTaskSessionImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/GridTaskSessionImpl.java
index 8478a9e..0bd1718 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridTaskSessionImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridTaskSessionImpl.java
@@ -114,9 +114,6 @@ public class GridTaskSessionImpl implements GridTaskSessionInternal {
private final IgnitePredicate<ClusterNode> topPred;
/** */
- private final UUID subjId;
-
- /** */
private final IgniteFutureImpl mapFut;
/** */
@@ -137,7 +134,6 @@ public class GridTaskSessionImpl implements GridTaskSessionInternal {
* @param ctx Grid Kernal Context.
* @param fullSup Session full support enabled flag.
* @param internal Internal task flag.
- * @param subjId Subject ID.
* @param execName Custom executor name.
*/
public GridTaskSessionImpl(
@@ -155,7 +151,6 @@ public class GridTaskSessionImpl implements GridTaskSessionInternal {
GridKernalContext ctx,
boolean fullSup,
boolean internal,
- UUID subjId,
@Nullable String execName) {
assert taskNodeId != null;
assert taskName != null;
@@ -185,7 +180,6 @@ public class GridTaskSessionImpl implements GridTaskSessionInternal {
this.fullSup = fullSup;
this.internal = internal;
- this.subjId = subjId;
this.execName = execName;
mapFut = new IgniteFutureImpl(new GridFutureAdapter());
@@ -196,11 +190,6 @@ public class GridTaskSessionImpl implements GridTaskSessionInternal {
return fullSup;
}
- /** {@inheritDoc} */
- @Override public UUID subjectId() {
- return subjId;
- }
-
/**
*
*/
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridTaskSessionInternal.java b/modules/core/src/main/java/org/apache/ignite/internal/GridTaskSessionInternal.java
index f00e93e..bc28ec0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridTaskSessionInternal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridTaskSessionInternal.java
@@ -17,7 +17,6 @@
package org.apache.ignite.internal;
-import java.util.UUID;
import org.apache.ignite.compute.ComputeTaskSession;
import org.apache.ignite.lang.IgniteUuid;
import org.jetbrains.annotations.Nullable;
@@ -60,9 +59,4 @@ public interface GridTaskSessionInternal extends ComputeTaskSession {
* @return {@code True} if checkpoints and attributes are enabled.
*/
public boolean isFullSupport();
-
- /**
- * @return Subject ID.
- */
- public UUID subjectId();
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteComputeImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteComputeImpl.java
index 60aa76a..ca9f534 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteComputeImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteComputeImpl.java
@@ -25,7 +25,6 @@ import java.io.ObjectStreamException;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
-import java.util.UUID;
import java.util.concurrent.Callable;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteCompute;
@@ -54,7 +53,6 @@ import static org.apache.ignite.internal.GridClosureCallMode.BROADCAST;
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_NO_FAILOVER;
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_NO_RESULT_CACHE;
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_SUBGRID_PREDICATE;
-import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_SUBJ_ID;
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_TASK_NAME;
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_TIMEOUT;
@@ -72,9 +70,6 @@ public class IgniteComputeImpl extends AsyncSupportAdapter<IgniteCompute>
/** */
private ClusterGroupAdapter prj;
- /** */
- private UUID subjId;
-
/** Custom executor name. */
private String execName;
@@ -88,24 +83,21 @@ public class IgniteComputeImpl extends AsyncSupportAdapter<IgniteCompute>
/**
* @param ctx Kernal context.
* @param prj Projection.
- * @param subjId Subject ID.
*/
- public IgniteComputeImpl(GridKernalContext ctx, ClusterGroupAdapter prj, UUID subjId) {
- this(ctx, prj, subjId, false);
+ public IgniteComputeImpl(GridKernalContext ctx, ClusterGroupAdapter prj) {
+ this(ctx, prj, false);
}
/**
* @param ctx Kernal context.
* @param prj Projection.
- * @param subjId Subject ID.
* @param async Async support flag.
*/
- private IgniteComputeImpl(GridKernalContext ctx, ClusterGroupAdapter prj, UUID subjId, boolean async) {
+ private IgniteComputeImpl(GridKernalContext ctx, ClusterGroupAdapter prj, boolean async) {
super(async);
this.ctx = ctx;
this.prj = prj;
- this.subjId = subjId;
}
/**
@@ -117,19 +109,18 @@ public class IgniteComputeImpl extends AsyncSupportAdapter<IgniteCompute>
* @param async Async support flag.
* @param execName Custom executor name.
*/
- private IgniteComputeImpl(GridKernalContext ctx, ClusterGroupAdapter prj, UUID subjId, boolean async,
+ private IgniteComputeImpl(GridKernalContext ctx, ClusterGroupAdapter prj, boolean async,
String execName) {
super(async);
this.ctx = ctx;
this.prj = prj;
- this.subjId = subjId;
this.execName = execName;
}
/** {@inheritDoc} */
@Override protected IgniteCompute createAsyncInstance() {
- return new IgniteComputeImpl(ctx, prj, subjId, true);
+ return new IgniteComputeImpl(ctx, prj, true);
}
/** {@inheritDoc} */
@@ -481,7 +472,6 @@ public class IgniteComputeImpl extends AsyncSupportAdapter<IgniteCompute>
try {
ctx.task().setThreadContextIfNotNull(TC_SUBGRID_PREDICATE, prj.predicate());
- ctx.task().setThreadContextIfNotNull(TC_SUBJ_ID, subjId);
return ctx.task().execute(taskName, arg, execName);
}
@@ -520,7 +510,6 @@ public class IgniteComputeImpl extends AsyncSupportAdapter<IgniteCompute>
try {
ctx.task().setThreadContextIfNotNull(TC_SUBGRID_PREDICATE, prj.predicate());
- ctx.task().setThreadContextIfNotNull(TC_SUBJ_ID, subjId);
return ctx.task().execute(taskCls, arg, execName);
}
@@ -559,7 +548,6 @@ public class IgniteComputeImpl extends AsyncSupportAdapter<IgniteCompute>
try {
ctx.task().setThreadContextIfNotNull(TC_SUBGRID_PREDICATE, prj.predicate());
- ctx.task().setThreadContextIfNotNull(TC_SUBJ_ID, subjId);
return ctx.task().execute(task, arg, execName);
}
@@ -1132,6 +1120,6 @@ public class IgniteComputeImpl extends AsyncSupportAdapter<IgniteCompute>
/** {@inheritDoc} */
@Override public IgniteCompute withExecutor(@NotNull String name) {
- return new IgniteComputeImpl(ctx, prj, subjId, isAsync(), name);
+ return new IgniteComputeImpl(ctx, prj, isAsync(), name);
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
index 50444be..ce027ec 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
@@ -1097,6 +1097,9 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
startProcessor(new FailureProcessor(ctx));
+ // Start security processors.
+ startProcessor(securityProcessor());
+
PoolProcessor pools = new PoolProcessor(ctx);
startProcessor(pools);
@@ -1116,9 +1119,6 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
// as managers may depend on it.
startProcessor(new GridTimeoutProcessor(ctx));
- // Start security processors.
- startProcessor(securityProcessor());
-
// Start SPI managers.
// NOTE: that order matters as there are dependencies between managers.
try {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java
index 22f4f04..f9d8903 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java
@@ -90,9 +90,6 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
/** Ignite instance name. */
private String igniteInstanceName;
- /** Subject ID. */
- protected UUID subjId;
-
/** Cluster group predicate. */
protected IgnitePredicate<ClusterNode> p;
@@ -110,17 +107,14 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
}
/**
- * @param subjId Subject ID.
* @param ctx Kernal context.
* @param p Predicate.
*/
protected ClusterGroupAdapter(@Nullable GridKernalContext ctx,
- @Nullable UUID subjId,
@Nullable IgnitePredicate<ClusterNode> p) {
if (ctx != null)
setKernalContext(ctx);
- this.subjId = subjId;
this.p = p;
ids = null;
@@ -128,37 +122,31 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
/**
* @param ctx Kernal context.
- * @param subjId Subject ID.
* @param ids Node IDs.
*/
protected ClusterGroupAdapter(@Nullable GridKernalContext ctx,
- @Nullable UUID subjId,
Set<UUID> ids) {
if (ctx != null)
setKernalContext(ctx);
assert ids != null;
- this.subjId = subjId;
this.ids = ids;
p = F.nodeForNodeIds(ids);
}
/**
- * @param subjId Subject ID.
* @param ctx Grid kernal context.
* @param p Predicate.
* @param ids Node IDs.
*/
private ClusterGroupAdapter(@Nullable GridKernalContext ctx,
- @Nullable UUID subjId,
@Nullable IgnitePredicate<ClusterNode> p,
Set<UUID> ids) {
if (ctx != null)
setKernalContext(ctx);
- this.subjId = subjId;
this.p = p;
this.ids = ids;
@@ -219,7 +207,7 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
if (compute == null) {
assert ctx != null;
- compute = new IgniteComputeImpl(ctx, this, subjId);
+ compute = new IgniteComputeImpl(ctx, this);
}
return compute;
@@ -382,7 +370,7 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
if (p != null)
ctx.resource().injectGeneric(p);
- return new ClusterGroupAdapter(ctx, subjId, this.p != null ? F.and(p, this.p) : p);
+ return new ClusterGroupAdapter(ctx, this.p != null ? F.and(p, this.p) : p);
}
catch (IgniteCheckedException e) {
throw U.convertException(e);
@@ -431,7 +419,7 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
nodeIds.add(node.id());
}
- return new ClusterGroupAdapter(ctx, subjId, nodeIds);
+ return new ClusterGroupAdapter(ctx, nodeIds);
}
finally {
unguard();
@@ -451,7 +439,7 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
if (contains(n))
nodeIds.add(n.id());
- return new ClusterGroupAdapter(ctx, subjId, nodeIds);
+ return new ClusterGroupAdapter(ctx, nodeIds);
}
finally {
unguard();
@@ -481,7 +469,7 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
nodeIds.add(id);
}
- return new ClusterGroupAdapter(ctx, subjId, nodeIds);
+ return new ClusterGroupAdapter(ctx, nodeIds);
}
finally {
unguard();
@@ -502,7 +490,7 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
nodeIds.add(id);
}
- return new ClusterGroupAdapter(ctx, subjId, nodeIds);
+ return new ClusterGroupAdapter(ctx, nodeIds);
}
finally {
unguard();
@@ -533,7 +521,7 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
nodeIds.add(id);
}
- return new ClusterGroupAdapter(ctx, subjId, nodeIds);
+ return new ClusterGroupAdapter(ctx, nodeIds);
}
finally {
unguard();
@@ -566,7 +554,7 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
nodeIds.add(id);
}
- return new ClusterGroupAdapter(ctx, subjId, nodeIds);
+ return new ClusterGroupAdapter(ctx, nodeIds);
}
finally {
unguard();
@@ -667,22 +655,6 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
return new AgeClusterGroup(this, false);
}
- /** {@inheritDoc} */
- @Override public ClusterGroupEx forSubjectId(UUID subjId) {
- if (subjId == null)
- return this;
-
- guard();
-
- try {
- return ids != null ? new ClusterGroupAdapter(ctx, subjId, ids) :
- new ClusterGroupAdapter(ctx, subjId, p);
- }
- finally {
- unguard();
- }
- }
-
/**
* @param n Node.
* @return Whether node belongs to this cluster group.
@@ -752,7 +724,6 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
/** {@inheritDoc} */
@Override public void writeExternal(ObjectOutput out) throws IOException {
U.writeString(out, igniteInstanceName);
- U.writeUuid(out, subjId);
out.writeBoolean(ids != null);
@@ -765,7 +736,6 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
/** {@inheritDoc} */
@Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
igniteInstanceName = U.readString(in);
- subjId = U.readUuid(in);
if (in.readBoolean())
ids = (Set<UUID>)in.readObject();
@@ -783,8 +753,8 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
try {
IgniteKernal g = IgnitionEx.localIgnite();
- return ids != null ? new ClusterGroupAdapter(g.context(), subjId, ids) :
- new ClusterGroupAdapter(g.context(), subjId, p);
+ return ids != null ? new ClusterGroupAdapter(g.context(), ids) :
+ new ClusterGroupAdapter(g.context(), p);
}
catch (IllegalStateException e) {
throw U.withCause(new InvalidObjectException(e.getMessage()), e);
@@ -992,7 +962,7 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
* @param isOldest Oldest flag.
*/
private AgeClusterGroup(ClusterGroupAdapter parent, boolean isOldest) {
- super(parent.ctx, parent.subjId, parent.p, parent.ids);
+ super(parent.ctx, parent.p, parent.ids);
this.isOldest = isOldest;
}
@@ -1049,7 +1019,7 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
if (p != null)
ctx.resource().injectGeneric(p);
- return new ClusterGroupAdapter(ctx, this.subjId, new GroupPredicate(this, p));
+ return new ClusterGroupAdapter(ctx, new GroupPredicate(this, p));
}
catch (IgniteCheckedException e) {
throw U.convertException(e);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupEx.java b/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupEx.java
index 6ebd7ad..c582528 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupEx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupEx.java
@@ -17,7 +17,6 @@
package org.apache.ignite.internal.cluster;
-import java.util.UUID;
import org.apache.ignite.cluster.ClusterGroup;
/**
@@ -25,14 +24,6 @@ import org.apache.ignite.cluster.ClusterGroup;
*/
public interface ClusterGroupEx extends ClusterGroup {
/**
- * Creates projection for specified subject ID.
- *
- * @param subjId Subject ID.
- * @return Cluster group.
- */
- public ClusterGroupEx forSubjectId(UUID subjId);
-
- /**
* @param cacheName Cache name.
* @param affNodes Flag to include affinity nodes.
* @param nearNodes Flag to include near nodes.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterImpl.java
index 309172b..c1a80a1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterImpl.java
@@ -132,7 +132,7 @@ public class IgniteClusterImpl extends ClusterGroupAdapter implements IgniteClus
* @param ctx Kernal context.
*/
public IgniteClusterImpl(GridKernalContext ctx) {
- super(ctx, null, (IgnitePredicate<ClusterNode>)null);
+ super(ctx, (IgnitePredicate<ClusterNode>)null);
cfg = ctx.config();
@@ -158,7 +158,7 @@ public class IgniteClusterImpl extends ClusterGroupAdapter implements IgniteClus
guard();
try {
- return new ClusterGroupAdapter(ctx, null, Collections.singleton(ctx.discovery().localNode().id()));
+ return new ClusterGroupAdapter(ctx, Collections.singleton(ctx.discovery().localNode().id()));
}
finally {
unguard();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
index da91b54..fb5592d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
@@ -806,6 +806,8 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
ctx.encryption().onLocalJoin();
+ ctx.security().onLocalJoin();
+
ctx.cluster().onLocalJoin();
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java
index 286a6d0..32ed1d8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java
@@ -50,8 +50,6 @@ import org.apache.ignite.internal.processors.GridProcessorAdapter;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.pool.PoolProcessor;
import org.apache.ignite.internal.processors.resource.GridNoImplicitInjection;
-import org.apache.ignite.internal.processors.security.OperationSecurityContext;
-import org.apache.ignite.internal.processors.security.SecurityContext;
import org.apache.ignite.internal.util.GridSpinReadWriteLock;
import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
@@ -962,11 +960,9 @@ public class GridClosureProcessor extends GridProcessorAdapter {
final GridWorkerFuture<R> fut = new GridWorkerFuture<>();
- final SecurityContext secCtx = ctx.security().securityContext();
-
GridWorker w = new GridWorker(ctx.igniteInstanceName(), "closure-proc-worker", log) {
@Override protected void body() {
- try (OperationSecurityContext s = ctx.security().withContext(secCtx)) {
+ try {
if (ldr != null)
fut.onDone(U.wrapThreadLoader(ldr, c));
else
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java
index 47b5439..679834b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java
@@ -113,6 +113,7 @@ import static org.apache.ignite.internal.processors.cache.distributed.dht.topolo
import static org.apache.ignite.internal.processors.metric.GridMetricManager.CPU_LOAD;
import static org.apache.ignite.internal.processors.metric.GridMetricManager.SYS_METRICS;
import static org.apache.ignite.internal.processors.metric.impl.MetricUtils.metricName;
+import static org.apache.ignite.internal.processors.security.SecurityUtils.securitySubjectId;
import static org.jsr166.ConcurrentLinkedHashMap.QueuePolicy.PER_SEGMENT_Q;
/**
@@ -1247,7 +1248,6 @@ public class GridJobProcessor extends GridProcessorAdapter {
sesAttrs,
req.isSessionFullSupport(),
req.isInternal(),
- req.getSubjectId(),
req.executorName());
taskSes.setCheckpointSpi(req.getCheckpointSpi());
@@ -1559,7 +1559,7 @@ public class GridJobProcessor extends GridProcessorAdapter {
evt.taskSessionId(req.getSessionId());
evt.type(EVT_JOB_FAILED);
evt.taskNode(node);
- evt.taskSubjectId(req.getSubjectId());
+ evt.taskSubjectId(securitySubjectId(ctx));
// Record job reply failure.
ctx.event().record(evt);
@@ -1637,7 +1637,7 @@ public class GridJobProcessor extends GridProcessorAdapter {
evt.taskSessionId(req.getSessionId());
evt.type(EVT_JOB_FAILED);
evt.taskNode(node);
- evt.taskSubjectId(req.getSubjectId());
+ evt.taskSubjectId(securitySubjectId(ctx));
// Record job reply failure.
ctx.event().record(evt);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobWorker.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobWorker.java
index fa8add3..f78be69 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobWorker.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobWorker.java
@@ -539,7 +539,7 @@ public class GridJobWorker extends GridWorker implements GridTimeoutObject {
SqlFieldsQuery.setThreadedQueryInitiatorId("task:" + ses.getTaskName() + ":" + getJobId());
- try {
+ try (OperationSecurityContext ignored = ctx.security().withContext(secCtx)) {
if (partsReservation != null) {
try {
if (!partsReservation.reserve()) {
@@ -802,7 +802,7 @@ public class GridJobWorker extends GridWorker implements GridTimeoutObject {
evt.taskSessionId(ses.getId());
evt.type(evtType);
evt.taskNode(taskNode);
- evt.taskSubjectId(ses.subjectId());
+ evt.taskSubjectId(secCtx != null ? secCtx.subject().id() : null);
ctx.event().record(evt);
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/compute/ClientComputeTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/compute/ClientComputeTask.java
index 4ce3907..711d7c7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/compute/ClientComputeTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/compute/ClientComputeTask.java
@@ -38,7 +38,6 @@ import static org.apache.ignite.internal.processors.platform.client.ClientMessag
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_NO_FAILOVER;
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_NO_RESULT_CACHE;
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_SUBGRID_PREDICATE;
-import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_SUBJ_ID;
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_TIMEOUT;
/**
@@ -100,10 +99,7 @@ class ClientComputeTask implements ClientCloseableResource {
IgnitePredicate<ClusterNode> nodePredicate = F.isEmpty(nodeIds) ? node -> !node.isClient() :
F.nodeForNodeIds(nodeIds);
- UUID subjId = ctx.securityContext() == null ? null : ctx.securityContext().subject().id();
-
task.setThreadContext(TC_SUBGRID_PREDICATE, nodePredicate);
- task.setThreadContextIfNotNull(TC_SUBJ_ID, subjId);
task.setThreadContext(TC_TIMEOUT, timeout);
task.setThreadContext(TC_NO_FAILOVER, (flags & NO_FAILOVER_FLAG_MASK) != 0);
task.setThreadContext(TC_NO_RESULT_CACHE, (flags & NO_RESULT_CACHE_FLAG_MASK) != 0);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/service/ClientServiceInvokeRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/service/ClientServiceInvokeRequest.java
index 456d077..7d4c0dc 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/service/ClientServiceInvokeRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/service/ClientServiceInvokeRequest.java
@@ -143,9 +143,6 @@ public class ClientServiceInvokeRequest extends ClientRequest {
ClusterGroupAdapter grp = ctx.kernalContext().cluster().get();
- if (ctx.securityContext() != null)
- grp = (ClusterGroupAdapter)grp.forSubjectId(ctx.securityContext().subject().id());
-
grp = (ClusterGroupAdapter)(nodeIds.isEmpty() ? grp.forServers() : grp.forNodeIds(nodeIds));
IgniteServices services = grp.services();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/pool/PoolProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/pool/PoolProcessor.java
index f8fe6ce..3a6e8b0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/pool/PoolProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/pool/PoolProcessor.java
@@ -24,6 +24,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
@@ -40,10 +41,16 @@ import org.apache.ignite.internal.IgniteComponentType;
import org.apache.ignite.internal.managers.communication.GridIoPolicy;
import org.apache.ignite.internal.processors.GridProcessorAdapter;
import org.apache.ignite.internal.processors.plugin.IgnitePluginProcessor;
+import org.apache.ignite.internal.processors.security.IgniteSecurity;
+import org.apache.ignite.internal.processors.security.thread.SecurityAwareIoPool;
+import org.apache.ignite.internal.processors.security.thread.SecurityAwareStripedExecutor;
+import org.apache.ignite.internal.processors.security.thread.SecurityAwareStripedThreadPoolExecutor;
+import org.apache.ignite.internal.processors.security.thread.SecurityAwareThreadPoolExecutor;
import org.apache.ignite.internal.util.StripedExecutor;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.internal.util.worker.GridWorkerListener;
import org.apache.ignite.internal.worker.WorkersRegistry;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.plugin.extensions.communication.IoPool;
@@ -165,7 +172,7 @@ public class PoolProcessor extends GridProcessorAdapter {
throw new IgniteException("Failed to register IO executor pool because its ID as " +
"already used: " + id);
- extPools[id] = ex;
+ extPools[id] = ctx.security().enabled() ? new SecurityAwareIoPool(ctx.security(), ex) : ex;
}
}
}
@@ -187,7 +194,7 @@ public class PoolProcessor extends GridProcessorAdapter {
validateThreadPoolSize(cfg.getPublicThreadPoolSize(), "public");
- execSvc = new IgniteThreadPoolExecutor(
+ execSvc = createExecutorService(
"pub",
cfg.getIgniteInstanceName(),
cfg.getPublicThreadPoolSize(),
@@ -201,7 +208,7 @@ public class PoolProcessor extends GridProcessorAdapter {
validateThreadPoolSize(cfg.getServiceThreadPoolSize(), "service");
- svcExecSvc = new IgniteThreadPoolExecutor(
+ svcExecSvc = createExecutorService(
"svc",
cfg.getIgniteInstanceName(),
cfg.getServiceThreadPoolSize(),
@@ -215,7 +222,7 @@ public class PoolProcessor extends GridProcessorAdapter {
validateThreadPoolSize(cfg.getSystemThreadPoolSize(), "system");
- sysExecSvc = new IgniteThreadPoolExecutor(
+ sysExecSvc = createExecutorService(
"sys",
cfg.getIgniteInstanceName(),
cfg.getSystemThreadPoolSize(),
@@ -231,7 +238,7 @@ public class PoolProcessor extends GridProcessorAdapter {
WorkersRegistry workerRegistry = ctx.workersRegistry();
- stripedExecSvc = new StripedExecutor(
+ stripedExecSvc = createStripedExecutor(
cfg.getStripedPoolSize(),
cfg.getIgniteInstanceName(),
"sys",
@@ -241,6 +248,7 @@ public class PoolProcessor extends GridProcessorAdapter {
ctx.failure().process(new FailureContext(SYSTEM_WORKER_TERMINATION, t));
}
},
+ false,
workerRegistry,
cfg.getFailureDetectionTimeout());
@@ -250,7 +258,7 @@ public class PoolProcessor extends GridProcessorAdapter {
// not be needed.
validateThreadPoolSize(cfg.getManagementThreadPoolSize(), "management");
- mgmtExecSvc = new IgniteThreadPoolExecutor(
+ mgmtExecSvc = createExecutorService(
"mgmt",
cfg.getIgniteInstanceName(),
cfg.getManagementThreadPoolSize(),
@@ -268,7 +276,7 @@ public class PoolProcessor extends GridProcessorAdapter {
// not be needed.
validateThreadPoolSize(cfg.getPeerClassLoadingThreadPoolSize(), "peer class loading");
- p2pExecSvc = new IgniteThreadPoolExecutor(
+ p2pExecSvc = createExecutorService(
"p2p",
cfg.getIgniteInstanceName(),
cfg.getPeerClassLoadingThreadPoolSize(),
@@ -280,7 +288,7 @@ public class PoolProcessor extends GridProcessorAdapter {
p2pExecSvc.allowCoreThreadTimeOut(true);
- dataStreamerExecSvc = new StripedExecutor(
+ dataStreamerExecSvc = createStripedExecutor(
cfg.getDataStreamerThreadPoolSize(),
cfg.getIgniteInstanceName(),
"data-streamer",
@@ -308,7 +316,7 @@ public class PoolProcessor extends GridProcessorAdapter {
if (cfg.getConnectorConfiguration() != null) {
validateThreadPoolSize(cfg.getConnectorConfiguration().getThreadPoolSize(), "connector");
- restExecSvc = new IgniteThreadPoolExecutor(
+ restExecSvc = createExecutorService(
"rest",
cfg.getIgniteInstanceName(),
cfg.getConnectorConfiguration().getThreadPoolSize(),
@@ -324,7 +332,7 @@ public class PoolProcessor extends GridProcessorAdapter {
validateThreadPoolSize(cfg.getUtilityCacheThreadPoolSize(), "utility cache");
- utilityCacheExecSvc = new IgniteThreadPoolExecutor(
+ utilityCacheExecSvc = createExecutorService(
"utility",
cfg.getIgniteInstanceName(),
cfg.getUtilityCacheThreadPoolSize(),
@@ -336,7 +344,7 @@ public class PoolProcessor extends GridProcessorAdapter {
utilityCacheExecSvc.allowCoreThreadTimeOut(true);
- affExecSvc = new IgniteThreadPoolExecutor(
+ affExecSvc = createExecutorService(
"aff",
cfg.getIgniteInstanceName(),
1,
@@ -351,7 +359,7 @@ public class PoolProcessor extends GridProcessorAdapter {
if (IgniteComponentType.INDEXING.inClassPath()) {
int cpus = Runtime.getRuntime().availableProcessors();
- idxExecSvc = new IgniteThreadPoolExecutor(
+ idxExecSvc = createExecutorService(
"idx",
cfg.getIgniteInstanceName(),
cpus,
@@ -366,7 +374,7 @@ public class PoolProcessor extends GridProcessorAdapter {
validateThreadPoolSize(buildIdxThreadPoolSize, "build-idx");
- buildIdxExecSvc = new IgniteThreadPoolExecutor(
+ buildIdxExecSvc = createExecutorService(
"build-idx-runner",
cfg.getIgniteInstanceName(),
buildIdxThreadPoolSize,
@@ -382,7 +390,7 @@ public class PoolProcessor extends GridProcessorAdapter {
validateThreadPoolSize(cfg.getQueryThreadPoolSize(), "query");
- qryExecSvc = new IgniteThreadPoolExecutor(
+ qryExecSvc = createExecutorService(
"query",
cfg.getIgniteInstanceName(),
cfg.getQueryThreadPoolSize(),
@@ -394,7 +402,7 @@ public class PoolProcessor extends GridProcessorAdapter {
qryExecSvc.allowCoreThreadTimeOut(true);
- schemaExecSvc = new IgniteThreadPoolExecutor(
+ schemaExecSvc = createExecutorService(
"schema",
cfg.getIgniteInstanceName(),
2,
@@ -408,7 +416,7 @@ public class PoolProcessor extends GridProcessorAdapter {
validateThreadPoolSize(cfg.getRebalanceThreadPoolSize(), "rebalance");
- rebalanceExecSvc = new IgniteThreadPoolExecutor(
+ rebalanceExecSvc = createExecutorService(
"rebalance",
cfg.getIgniteInstanceName(),
cfg.getRebalanceThreadPoolSize(),
@@ -420,7 +428,7 @@ public class PoolProcessor extends GridProcessorAdapter {
rebalanceExecSvc.allowCoreThreadTimeOut(true);
- rebalanceStripedExecSvc = new IgniteStripedThreadPoolExecutor(
+ rebalanceStripedExecSvc = createStripedThreadPoolExecutor(
cfg.getRebalanceThreadPoolSize(),
cfg.getIgniteInstanceName(),
"rebalance-striped",
@@ -434,7 +442,7 @@ public class PoolProcessor extends GridProcessorAdapter {
customExecs = new HashMap<>();
for (ExecutorConfiguration execCfg : cfg.getExecutorConfiguration()) {
- ThreadPoolExecutor exec = new IgniteThreadPoolExecutor(
+ ThreadPoolExecutor exec = createExecutorService(
execCfg.getName(),
cfg.getIgniteInstanceName(),
execCfg.getSize(),
@@ -851,4 +859,89 @@ public class PoolProcessor extends GridProcessorAdapter {
", size=" + cfg.getSize() + ']');
}
}
+
+ /** Creates instance {@link IgniteStripedThreadPoolExecutor} with a notion of whether {@link IgniteSecurity} is enabled. */
+ private IgniteStripedThreadPoolExecutor createStripedThreadPoolExecutor(
+ int concurrentLvl,
+ String igniteInstanceName,
+ String threadNamePrefix,
+ UncaughtExceptionHandler eHnd,
+ boolean allowCoreThreadTimeOut,
+ long keepAliveTime
+ ) {
+ return ctx.security().enabled()
+ ? new SecurityAwareStripedThreadPoolExecutor(
+ ctx.security(),
+ concurrentLvl,
+ igniteInstanceName,
+ threadNamePrefix,
+ eHnd,
+ allowCoreThreadTimeOut,
+ keepAliveTime)
+ : new IgniteStripedThreadPoolExecutor(
+ concurrentLvl,
+ igniteInstanceName,
+ threadNamePrefix,
+ eHnd,
+ allowCoreThreadTimeOut,
+ keepAliveTime);
+ }
+
+ /** Creates instance {@link StripedExecutor} with a notion of whether {@link IgniteSecurity} is enabled. */
+ private StripedExecutor createStripedExecutor(
+ int cnt,
+ String igniteInstanceName,
+ String poolName,
+ final IgniteLogger log,
+ IgniteInClosure<Throwable> errHnd,
+ boolean stealTasks,
+ GridWorkerListener gridWorkerLsnr,
+ long failureDetectionTimeout
+ ) {
+ return ctx.security().enabled()
+ ? new SecurityAwareStripedExecutor(
+ ctx.security(),
+ cnt,
+ igniteInstanceName,
+ poolName,
+ log,
+ errHnd,
+ stealTasks,
+ gridWorkerLsnr,
+ failureDetectionTimeout)
+ : new StripedExecutor(cnt, igniteInstanceName, poolName, log, errHnd, stealTasks, gridWorkerLsnr, failureDetectionTimeout);
+ }
+
+ /** Creates instance {@link IgniteThreadPoolExecutor} with a notion of whether {@link IgniteSecurity} is enabled. */
+ private IgniteThreadPoolExecutor createExecutorService(
+ String threadNamePrefix,
+ String igniteInstanceName,
+ int corePoolSize,
+ int maxPoolSize,
+ long keepAliveTime,
+ BlockingQueue<Runnable> workQ,
+ byte plc,
+ UncaughtExceptionHandler eHnd
+ ) {
+ return ctx.security().enabled()
+ ? new SecurityAwareThreadPoolExecutor(
+ ctx.security(),
+ threadNamePrefix,
+ igniteInstanceName,
+ corePoolSize,
+ maxPoolSize,
+ keepAliveTime,
+ workQ,
+ plc,
+ eHnd)
+ : new IgniteThreadPoolExecutor(
+ threadNamePrefix,
+ igniteInstanceName,
+ corePoolSize,
+ maxPoolSize,
+ keepAliveTime,
+ workQ,
+ plc,
+ eHnd);
+ }
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/GridRestProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/GridRestProcessor.java
index f774c8d..ae59f6d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/GridRestProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/GridRestProcessor.java
@@ -145,6 +145,9 @@ public class GridRestProcessor extends GridProcessorAdapter implements IgniteRes
/** */
private final Thread sesTimeoutCheckerThread;
+ /** */
+ private final boolean securityEnabled;
+
/** Protocol handler. */
private final GridRestProtocolHandler protoHnd = new GridRestProtocolHandler() {
@Override public GridRestResponse handle(GridRestRequest req) throws IgniteCheckedException {
@@ -249,8 +252,6 @@ public class GridRestProcessor extends GridProcessorAdapter implements IgniteRes
if (log.isDebugEnabled())
log.debug("Received request from client: " + req);
- boolean securityEnabled = ctx.security().enabled();
-
if (securityEnabled) {
Session ses;
@@ -281,9 +282,9 @@ public class GridRestProcessor extends GridProcessorAdapter implements IgniteRes
try (OperationSecurityContext s = ctx.security().withContext(secCtx0)) {
authorize(req);
- }
- req.securityContext(secCtx0);
+ return handleRequest0(req);
+ }
}
catch (SecurityException e) {
assert secCtx0 != null;
@@ -294,7 +295,17 @@ public class GridRestProcessor extends GridProcessorAdapter implements IgniteRes
return new GridFinishedFuture<>(new GridRestResponse(STATUS_AUTH_FAILED, e.getMessage()));
}
}
+ else
+ return handleRequest0(req);
+ }
+ /**
+ * Performs request handling.
+ *
+ * @param req Request to handle.
+ * @return Future of request execution.
+ */
+ private IgniteInternalFuture<GridRestResponse> handleRequest0(GridRestRequest req) {
interceptRequest(req);
GridRestCommandHandler hnd = handlers.get(req.command());
@@ -390,7 +401,7 @@ public class GridRestProcessor extends GridProcessorAdapter implements IgniteRes
while (true) {
if (F.isEmpty(sesTok) && clientId == null) {
// TODO: In IGNITE 3.0 we should check credentials only for AUTHENTICATE command.
- if (ctx.security().enabled() && req.command() != AUTHENTICATE && req.credentials() == null)
+ if (securityEnabled && req.command() != AUTHENTICATE && req.credentials() == null)
throw new IgniteAuthenticationException("Failed to handle request - session token not found or invalid");
Session ses = Session.random();
@@ -479,6 +490,8 @@ public class GridRestProcessor extends GridProcessorAdapter implements IgniteRes
sesTtl = IgniteSystemProperties.getLong(IGNITE_REST_SESSION_TIMEOUT, DFLT_SES_TIMEOUT) * 1000;
sesTokTtl = IgniteSystemProperties.getLong(IGNITE_REST_SECURITY_TOKEN_TIMEOUT, DFLT_SES_TOKEN_INVALIDATE_INTERVAL) * 1000;
+ securityEnabled = ctx.security().enabled();
+
sesTimeoutCheckerThread = new IgniteThread(ctx.igniteInstanceName(), "session-timeout-worker",
new GridWorker(ctx.igniteInstanceName(), "session-timeout-worker", log) {
@Override protected void body() throws InterruptedException {
@@ -492,7 +505,7 @@ public class GridRestProcessor extends GridProcessorAdapter implements IgniteRes
clientId2SesId.remove(ses.clientId, ses.sesId);
sesId2Ses.remove(ses.sesId, ses);
- if (ctx.security().enabled() && ses.secCtx != null && ses.secCtx.subject() != null)
+ if (securityEnabled && ses.secCtx != null && ses.secCtx.subject() != null)
ctx.security().onSessionExpired(ses.secCtx.subject().id());
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/task/GridTaskCommandHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/task/GridTaskCommandHandler.java
index a689151..19d8af1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/task/GridTaskCommandHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/task/GridTaskCommandHandler.java
@@ -78,7 +78,6 @@ import static org.apache.ignite.internal.processors.rest.GridRestCommand.EXE;
import static org.apache.ignite.internal.processors.rest.GridRestCommand.NOOP;
import static org.apache.ignite.internal.processors.rest.GridRestCommand.RESULT;
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_NO_FAILOVER;
-import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_SUBJ_ID;
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_TIMEOUT;
import static org.jsr166.ConcurrentLinkedHashMap.QueuePolicy.PER_SEGMENT_Q;
@@ -214,12 +213,9 @@ public class GridTaskCommandHandler extends GridRestCommandHandlerAdapter {
long timeout = req0.timeout();
- final UUID clientId = req.clientId();
-
final IgniteInternalFuture<Object> taskFut;
if (locExec) {
- ctx.task().setThreadContextIfNotNull(TC_SUBJ_ID, clientId);
ctx.task().setThreadContext(TC_TIMEOUT, timeout);
Object arg = !F.isEmpty(params) ? params.size() == 1 ? params.get(0) : params.toArray() : null;
@@ -235,7 +231,7 @@ public class GridTaskCommandHandler extends GridRestCommandHandlerAdapter {
taskFut = ctx.closure().callAsync(
BALANCE,
- new ExeCallable(name, params, timeout, clientId), prj.nodes());
+ new ExeCallable(name, params, timeout), prj.nodes());
}
if (async) {
@@ -615,9 +611,6 @@ public class GridTaskCommandHandler extends GridRestCommandHandlerAdapter {
private long timeout;
/** */
- private UUID clientId;
-
- /** */
@IgniteInstanceResource
private IgniteEx g;
@@ -632,19 +625,16 @@ public class GridTaskCommandHandler extends GridRestCommandHandlerAdapter {
* @param name Name.
* @param params Params.
* @param timeout Timeout.
- * @param clientId Client ID.
*/
- private ExeCallable(String name, List<Object> params, long timeout, UUID clientId) {
+ private ExeCallable(String name, List<Object> params, long timeout) {
this.name = name;
this.params = params;
this.timeout = timeout;
- this.clientId = clientId;
}
/** {@inheritDoc} */
@Override public Object call() throws Exception {
- return g.compute(g.cluster().forSubjectId(clientId)).execute(
- name,
+ return g.compute().execute(name,
!params.isEmpty() ? params.size() == 1 ? params.get(0) : params.toArray() : null);
}
@@ -653,7 +643,6 @@ public class GridTaskCommandHandler extends GridRestCommandHandlerAdapter {
U.writeString(out, name);
out.writeObject(params);
out.writeLong(timeout);
- U.writeUuid(out, clientId);
}
/** {@inheritDoc} */
@@ -661,7 +650,6 @@ public class GridTaskCommandHandler extends GridRestCommandHandlerAdapter {
name = U.readString(in);
params = (List<Object>)in.readObject();
timeout = in.readLong();
- clientId = U.readUuid(in);
}
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/user/UserActionCommandHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/user/UserActionCommandHandler.java
index 04ed963..bb4341c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/user/UserActionCommandHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/user/UserActionCommandHandler.java
@@ -66,8 +66,6 @@ public class UserActionCommandHandler extends GridRestCommandHandlerAdapter {
IgniteSecurity security = ctx.security();
- security.withContext(req.securityContext());
-
switch (cmd) {
case ADD_USER:
security.createUser(req0.user(), req0.password().toCharArray());
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/request/GridRestRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/request/GridRestRequest.java
index 91278a6..b484d22 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/request/GridRestRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/request/GridRestRequest.java
@@ -22,10 +22,8 @@ import java.security.cert.Certificate;
import java.util.Map;
import java.util.UUID;
import org.apache.ignite.internal.processors.rest.GridRestCommand;
-import org.apache.ignite.internal.processors.security.SecurityContext;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.internal.S;
-import org.jetbrains.annotations.Nullable;
/**
* Grid command request.
@@ -50,9 +48,6 @@ public class GridRestRequest {
/** Command. */
private GridRestCommand cmd;
- /** */
- private SecurityContext secCtx;
-
/** User attributes. */
Map<String, String> userAttrs;
@@ -156,20 +151,6 @@ public class GridRestRequest {
}
/**
- * @return Security context.
- */
- @Nullable public SecurityContext securityContext() {
- return secCtx;
- }
-
- /**
- * @param secCtx Security context.
- */
- public void securityContext(SecurityContext secCtx) {
- this.secCtx = secCtx;
- }
-
- /**
* Gets user attributes.
*
* @return User attributes.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/security/IgniteSecurity.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/IgniteSecurity.java
index ae97df4..73f294d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/security/IgniteSecurity.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/IgniteSecurity.java
@@ -152,4 +152,11 @@ public interface IgniteSecurity {
* @throws IgniteCheckedException If error occurred.
*/
public void dropUser(String login) throws IgniteCheckedException;
+
+ /**
+ * Callback for local join events for which the regular events are not generated.
+ * <p/>
+ * Local join event is expected in cases of joining to topology or client reconnect.
+ */
+ public void onLocalJoin();
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/security/IgniteSecurityProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/IgniteSecurityProcessor.java
index e3dcc70..0f7814d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/security/IgniteSecurityProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/IgniteSecurityProcessor.java
@@ -25,14 +25,17 @@ import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.NodeStoppingException;
import org.apache.ignite.internal.processors.GridProcessor;
import org.apache.ignite.internal.processors.security.sandbox.AccessControllerSandbox;
import org.apache.ignite.internal.processors.security.sandbox.IgniteSandbox;
import org.apache.ignite.internal.processors.security.sandbox.NoOpSandbox;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteFuture;
@@ -105,6 +108,9 @@ public class IgniteSecurityProcessor implements IgniteSecurity, GridProcessor {
/** Instance of IgniteSandbox. */
private IgniteSandbox sandbox;
+ /** Node local security context ready future. */
+ private final GridFutureAdapter<SecurityContext> nodeSecCtxReadyFut = new GridFutureAdapter<>();
+
/**
* @param ctx Grid kernal context.
* @param secPrc Security processor.
@@ -293,6 +299,11 @@ public class IgniteSecurityProcessor implements IgniteSecurity, GridProcessor {
/** {@inheritDoc} */
@Override public void onKernalStop(boolean cancel) {
+ if (!nodeSecCtxReadyFut.isDone()) {
+ nodeSecCtxReadyFut.onDone(new NodeStoppingException(
+ "Failed to wait for local node security context initialization (grid is stopping)."));
+ }
+
secPrc.onKernalStop(cancel);
}
@@ -367,13 +378,39 @@ public class IgniteSecurityProcessor implements IgniteSecurity, GridProcessor {
secPrc.dropUser(login);
}
+ /** {@inheritDoc} */
+ @Override public void onLocalJoin() {
+ try {
+ SecurityContext secCtx = nodeSecurityContext(
+ marsh,
+ U.resolveClassLoader(ctx.config()),
+ ctx.discovery().localNode());
+
+ nodeSecCtxReadyFut.onDone(secCtx);
+ }
+ catch (Throwable e) {
+ nodeSecCtxReadyFut.onDone(e);
+
+ throw e;
+ }
+ }
+
/**
* Getting local node's security context.
*
* @return Security context of local node.
*/
private SecurityContext localSecurityContext() {
- return nodeSecurityContext(marsh, U.resolveClassLoader(ctx.config()), ctx.discovery().localNode());
+ if (nodeSecCtxReadyFut.isDone()) {
+ try {
+ return nodeSecCtxReadyFut.get();
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+ }
+
+ return new DeferredSecurityContext(nodeSecCtxReadyFut);
}
/**
@@ -399,4 +436,54 @@ public class IgniteSecurityProcessor implements IgniteSecurity, GridProcessor {
public GridSecurityProcessor securityProcessor() {
return secPrc;
}
+
+ /**
+ * Represents {@link SecurityContext} wrapper that blocks all interface methods until security context becomes
+ * available. The main reason of such implementation is that local node security context is undefined until node
+ * joins the topology.
+ */
+ public static class DeferredSecurityContext implements SecurityContext {
+ /** */
+ private final GridFutureAdapter<SecurityContext> fut;
+
+ /** */
+ public DeferredSecurityContext(GridFutureAdapter<SecurityContext> fut) {
+ this.fut = fut;
+ }
+
+ /** {@inheritDoc} */
+ @Override public SecuritySubject subject() {
+ return delegate().subject();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean taskOperationAllowed(String taskClsName, SecurityPermission perm) {
+ return delegate().taskOperationAllowed(taskClsName, perm);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean cacheOperationAllowed(String cacheName, SecurityPermission perm) {
+ return delegate().cacheOperationAllowed(cacheName, perm);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean serviceOperationAllowed(String srvcName, SecurityPermission perm) {
+ return delegate().serviceOperationAllowed(srvcName, perm);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean systemOperationAllowed(SecurityPermission perm) {
+ return delegate().systemOperationAllowed(perm);
+ }
+
+ /** */
+ public SecurityContext delegate() {
+ try {
+ return fut.get();
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+ }
+ }
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/security/NoOpIgniteSecurityProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/NoOpIgniteSecurityProcessor.java
index db08f93..8205fc2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/security/NoOpIgniteSecurityProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/NoOpIgniteSecurityProcessor.java
@@ -165,4 +165,9 @@ public class NoOpIgniteSecurityProcessor extends GridProcessorAdapter implements
@Override public void dropUser(String login) throws IgniteCheckedException {
throw new IgniteException(SECURITY_DISABLED_ERROR_MSG);
}
+
+ /** {@inheritDoc} */
+ @Override public void onLocalJoin() {
+ // No-op.
+ }
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/security/SecurityUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/SecurityUtils.java
index 10970a6..d9c010f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/security/SecurityUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/SecurityUtils.java
@@ -35,6 +35,7 @@ import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
+import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -164,6 +165,13 @@ public class SecurityUtils {
}
}
+ /** @return Current security subject ID if security is enabled, otherwise null. */
+ public static UUID securitySubjectId(GridKernalContext ctx) {
+ IgniteSecurity security = ctx.security();
+
+ return security.enabled() ? security.securityContext().subject().id() : null;
+ }
+
/**
* Computes a result in a privileged action.
*
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/security/thread/SecurityAwareCallable.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/thread/SecurityAwareCallable.java
new file mode 100644
index 0000000..049403b
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/thread/SecurityAwareCallable.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.ignite.internal.processors.security.thread;
+
+import java.util.Collection;
+import java.util.concurrent.Callable;
+import java.util.stream.Collectors;
+import org.apache.ignite.internal.processors.security.IgniteSecurity;
+import org.apache.ignite.internal.processors.security.OperationSecurityContext;
+import org.apache.ignite.internal.processors.security.SecurityContext;
+
+/**
+ * Represents a {@link Callable} wrapper that executes the original {@link Callable} with the security context
+ * current at the time the wrapper was created.
+ */
+public class SecurityAwareCallable<T> implements Callable<T> {
+ /** Original callable. */
+ private final Callable<T> delegate;
+
+ /** */
+ private final IgniteSecurity security;
+
+ /** */
+ private final SecurityContext secCtx;
+
+ /** */
+ public SecurityAwareCallable(IgniteSecurity security, Callable<T> delegate) {
+ assert security.enabled();
+ assert delegate != null;
+
+ this.delegate = delegate;
+ this.security = security;
+ secCtx = security.securityContext();
+ }
+
+ /** {@inheritDoc} */
+ @Override public T call() throws Exception {
+ try (OperationSecurityContext ignored = security.withContext(secCtx)) {
+ return delegate.call();
+ }
+ }
+
+ /** */
+ public static <A> Collection<? extends Callable<A>> toSecurityAware(
+ IgniteSecurity sec,
+ Collection<? extends Callable<A>> tasks
+ ) {
+ return tasks.stream().map(t -> new SecurityAwareCallable<>(sec, t)).collect(Collectors.toList());
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/security/thread/SecurityAwareIoPool.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/thread/SecurityAwareIoPool.java
new file mode 100644
index 0000000..79b0440
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/thread/SecurityAwareIoPool.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.ignite.internal.processors.security.thread;
+
+import java.util.concurrent.Executor;
+import org.apache.ignite.internal.processors.security.IgniteSecurity;
+import org.apache.ignite.plugin.extensions.communication.IoPool;
+import org.jetbrains.annotations.NotNull;
+
+/** Wrapper of {@link IoPool} that executes tasks in security context that was actual when task was added to pool queue. */
+public class SecurityAwareIoPool implements IoPool {
+ /** */
+ private final IgniteSecurity security;
+
+ /** */
+ private final IoPool delegate;
+
+ /** */
+ private final Executor executor;
+
+ /** */
+ public SecurityAwareIoPool(IgniteSecurity security, IoPool delegate) {
+ assert security.enabled();
+ assert delegate != null;
+
+ this.security = security;
+ this.delegate = delegate;
+
+ final Executor delegateExecutor = delegate.executor();
+
+ executor = delegateExecutor == null ? null : new Executor() {
+ @Override public void execute(@NotNull Runnable cmd) {
+ delegateExecutor.execute(new SecurityAwareRunnable(SecurityAwareIoPool.this.security, cmd));
+ }
+ };
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte id() {
+ return delegate.id();
+ }
+
+ /** {@inheritDoc} */
+ @Override public Executor executor() {
+ return executor;
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/security/thread/SecurityAwareRunnable.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/thread/SecurityAwareRunnable.java
new file mode 100644
index 0000000..a5cb861
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/thread/SecurityAwareRunnable.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.ignite.internal.processors.security.thread;
+
+import org.apache.ignite.internal.processors.security.IgniteSecurity;
+import org.apache.ignite.internal.processors.security.OperationSecurityContext;
+import org.apache.ignite.internal.processors.security.SecurityContext;
+
+/**
+ * Represents a {@link Runnable} wrapper that executes the original {@link Runnable} with the security context
+ * current at the time the wrapper was created.
+ */
+public class SecurityAwareRunnable implements Runnable {
+ /** */
+ private final Runnable delegate;
+
+ /** */
+ private final IgniteSecurity security;
+
+ /** */
+ private final SecurityContext secCtx;
+
+ /** */
+ public SecurityAwareRunnable(IgniteSecurity security, Runnable delegate) {
+ assert security.enabled();
+ assert delegate != null;
+
+ this.delegate = delegate;
+ this.security = security;
+ secCtx = security.securityContext();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void run() {
+ try (OperationSecurityContext ignored = security.withContext(secCtx)) {
+ delegate.run();
+ }
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/security/thread/SecurityAwareStripedExecutor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/thread/SecurityAwareStripedExecutor.java
new file mode 100644
index 0000000..0420502
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/thread/SecurityAwareStripedExecutor.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.ignite.internal.processors.security.thread;
+
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.internal.processors.security.IgniteSecurity;
+import org.apache.ignite.internal.util.StripedExecutor;
+import org.apache.ignite.internal.util.worker.GridWorkerListener;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Extends {@link StripedExecutor} with the ability to execute tasks in security context that was actual when task was
+ * added to executor's queue.
+ */
+public class SecurityAwareStripedExecutor extends StripedExecutor {
+ /** */
+ private final IgniteSecurity security;
+
+ /** */
+ public SecurityAwareStripedExecutor(
+ IgniteSecurity security,
+ int cnt,
+ String igniteInstanceName,
+ String poolName,
+ IgniteLogger log,
+ IgniteInClosure<Throwable> errHnd,
+ GridWorkerListener gridWorkerLsnr,
+ long failureDetectionTimeout
+ ) {
+ super(cnt, igniteInstanceName, poolName, log, errHnd, gridWorkerLsnr, failureDetectionTimeout);
+
+ this.security = security;
+ }
+
+ /** */
+ public SecurityAwareStripedExecutor(
+ IgniteSecurity security,
+ int cnt,
+ String igniteInstanceName,
+ String poolName,
+ IgniteLogger log,
+ IgniteInClosure<Throwable> errHnd,
+ boolean stealTasks,
+ GridWorkerListener gridWorkerLsnr,
+ long failureDetectionTimeout
+ ) {
+ super(cnt, igniteInstanceName, poolName, log, errHnd, stealTasks, gridWorkerLsnr, failureDetectionTimeout);
+
+ this.security = security;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void execute(int idx, Runnable cmd) {
+ super.execute(idx, cmd == null ? null : new SecurityAwareRunnable(security, cmd));
+ }
+
+ /** {@inheritDoc} */
+ @Override public void execute(@NotNull Runnable cmd) {
+ super.execute(cmd == null ? null : new SecurityAwareRunnable(security, cmd));
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/security/thread/SecurityAwareStripedThreadPoolExecutor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/thread/SecurityAwareStripedThreadPoolExecutor.java
new file mode 100644
index 0000000..283a736
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/thread/SecurityAwareStripedThreadPoolExecutor.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.ignite.internal.processors.security.thread;
+
+import org.apache.ignite.internal.processors.security.IgniteSecurity;
+import org.apache.ignite.thread.IgniteStripedThreadPoolExecutor;
+
+/**
+ * Extends {@link IgniteStripedThreadPoolExecutor} with the ability to execute tasks in security context that was actual
+ * when task was added to executor's queue.
+ */
+public class SecurityAwareStripedThreadPoolExecutor extends IgniteStripedThreadPoolExecutor {
+ /** */
+ private final IgniteSecurity security;
+
+ /** */
+ public SecurityAwareStripedThreadPoolExecutor(
+ IgniteSecurity security,
+ int concurrentLvl,
+ String igniteInstanceName,
+ String threadNamePrefix,
+ Thread.UncaughtExceptionHandler eHnd,
+ boolean allowCoreThreadTimeOut,
+ long keepAliveTime
+ ) {
+ super(concurrentLvl, igniteInstanceName, threadNamePrefix, eHnd, allowCoreThreadTimeOut, keepAliveTime);
+ this.security = security;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void execute(Runnable task, int idx) {
+ super.execute(task == null ? null : new SecurityAwareRunnable(security, task), idx);
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/security/thread/SecurityAwareThreadPoolExecutor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/thread/SecurityAwareThreadPoolExecutor.java
new file mode 100644
index 0000000..7c77575
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/thread/SecurityAwareThreadPoolExecutor.java
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.ignite.internal.processors.security.thread;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.apache.ignite.internal.processors.security.IgniteSecurity;
+import org.apache.ignite.thread.IgniteThreadPoolExecutor;
+import org.jetbrains.annotations.NotNull;
+
+import static org.apache.ignite.internal.processors.security.thread.SecurityAwareCallable.toSecurityAware;
+
+/**
+ * Extends {@link ThreadPoolExecutor} with the ability to execute tasks in security context that was actual when task was
+ * added to executor's queue.
+ */
+public class SecurityAwareThreadPoolExecutor extends IgniteThreadPoolExecutor {
+ /** */
+ private final IgniteSecurity security;
+
+ /** */
+ public SecurityAwareThreadPoolExecutor(
+ IgniteSecurity security,
+ String threadNamePrefix,
+ String igniteInstanceName,
+ int corePoolSize,
+ int maxPoolSize,
+ long keepAliveTime,
+ BlockingQueue<Runnable> workQ,
+ byte plc,
+ Thread.UncaughtExceptionHandler eHnd
+ ) {
+ super(threadNamePrefix, igniteInstanceName, corePoolSize, maxPoolSize, keepAliveTime, workQ, plc, eHnd);
+
+ this.security = security;
+ }
+
+ /** {@inheritDoc} */
+ @NotNull @Override public <T> Future<T> submit(@NotNull Callable<T> task) {
+ return super.submit(task == null ? null : new SecurityAwareCallable<>(security, task));
+ }
+
+ /** {@inheritDoc} */
+ @NotNull @Override public <T> Future<T> submit(@NotNull Runnable task, T res) {
+ return super.submit(task == null ? null : new SecurityAwareRunnable(security, task), res);
+ }
+
+ /** {@inheritDoc} */
+ @NotNull @Override public Future<?> submit(@NotNull Runnable task) {
+ return super.submit(task == null ? null : new SecurityAwareRunnable(security, task));
+ }
+
+ /** {@inheritDoc} */
+ @NotNull @Override public <T> List<Future<T>> invokeAll(
+ @NotNull Collection<? extends Callable<T>> tasks) throws InterruptedException {
+ return super.invokeAll(tasks == null ? null : toSecurityAware(security, tasks));
+ }
+
+ /** {@inheritDoc} */
+ @NotNull @Override public <T> List<Future<T>> invokeAll(@NotNull Collection<? extends Callable<T>> tasks,
+ long timeout, @NotNull TimeUnit unit) throws InterruptedException {
+ return super.invokeAll(tasks == null ? null : toSecurityAware(security, tasks), timeout, unit);
+ }
+
+ /** {@inheritDoc} */
+ @NotNull @Override public <T> T invokeAny(@NotNull Collection<? extends Callable<T>> tasks)
+ throws InterruptedException, ExecutionException {
+ return super.invokeAny(tasks == null ? null : toSecurityAware(security, tasks));
+ }
+
+ /** {@inheritDoc} */
+ @Override public <T> T invokeAny(@NotNull Collection<? extends Callable<T>> tasks,
+ long timeout, @NotNull TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
+ return super.invokeAny(tasks == null ? null : toSecurityAware(security, tasks), timeout, unit);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void execute(@NotNull Runnable cmd) {
+ super.execute(cmd == null ? null : new SecurityAwareRunnable(security, cmd));
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/session/GridTaskSessionProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/session/GridTaskSessionProcessor.java
index 02b36eb..3feeb65 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/session/GridTaskSessionProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/session/GridTaskSessionProcessor.java
@@ -78,7 +78,6 @@ public class GridTaskSessionProcessor extends GridProcessorAdapter {
* @param attrs Map of attributes.
* @param fullSup {@code True} to enable distributed session attributes and checkpoints.
* @param internal {@code True} in case of internal task.
- * @param subjId Subject ID.
* @param execName Custom executor name.
* @return New session if one did not exist, or existing one.
*/
@@ -96,7 +95,6 @@ public class GridTaskSessionProcessor extends GridProcessorAdapter {
Map<Object, Object> attrs,
boolean fullSup,
boolean internal,
- UUID subjId,
@Nullable String execName) {
if (!fullSup) {
return new GridTaskSessionImpl(
@@ -114,7 +112,6 @@ public class GridTaskSessionProcessor extends GridProcessorAdapter {
ctx,
false,
internal,
- subjId,
execName);
}
@@ -139,7 +136,6 @@ public class GridTaskSessionProcessor extends GridProcessorAdapter {
ctx,
true,
internal,
- subjId,
execName));
if (old != null)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskProcessor.java
index 1bb6974..4cbbf9a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskProcessor.java
@@ -96,10 +96,10 @@ import static org.apache.ignite.internal.GridTopic.TOPIC_TASK_CANCEL;
import static org.apache.ignite.internal.managers.communication.GridIoPolicy.SYSTEM_POOL;
import static org.apache.ignite.internal.processors.cache.GridCacheUtils.isPersistenceEnabled;
import static org.apache.ignite.internal.processors.metric.GridMetricManager.SYS_METRICS;
+import static org.apache.ignite.internal.processors.security.SecurityUtils.securitySubjectId;
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_SKIP_AUTH;
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_SUBGRID;
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_SUBGRID_PREDICATE;
-import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_SUBJ_ID;
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_TASK_NAME;
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_TIMEOUT;
@@ -737,14 +737,6 @@ public class GridTaskProcessor extends GridProcessorAdapter implements IgniteCha
top = nodes != null ? F.nodeIds(nodes) : null;
}
- UUID subjId = (UUID)map.get(TC_SUBJ_ID);
-
- if (subjId == null)
- subjId = getThreadContext(TC_SUBJ_ID);
-
- if (subjId == null)
- subjId = ctx.localNodeId();
-
boolean internal = false;
if (dep == null || taskCls == null)
@@ -767,7 +759,6 @@ public class GridTaskProcessor extends GridProcessorAdapter implements IgniteCha
Collections.emptyMap(),
fullSup,
internal,
- subjId,
execName);
ComputeTaskInternalFuture<R> fut = new ComputeTaskInternalFuture<>(ses, ctx);
@@ -797,7 +788,7 @@ public class GridTaskProcessor extends GridProcessorAdapter implements IgniteCha
dep,
new TaskEventListener(),
map,
- subjId);
+ securitySubjectId(ctx));
GridTaskWorker<?, ?> taskWorker0 = tasks.putIfAbsent(sesId, taskWorker);
@@ -815,7 +806,7 @@ public class GridTaskProcessor extends GridProcessorAdapter implements IgniteCha
taskCls == null ? null : taskCls.getSimpleName(),
"VisorManagementTask",
false,
- subjId
+ securitySubjectId(ctx)
);
ctx.event().record(evt);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskThreadContextKey.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskThreadContextKey.java
index bffcecd..59249ec 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskThreadContextKey.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskThreadContextKey.java
@@ -39,9 +39,6 @@ public enum GridTaskThreadContextKey {
/** Timeout in milliseconds associated with the task. */
TC_TIMEOUT,
- /** Security subject ID. */
- TC_SUBJ_ID,
-
/** IO manager policy. */
TC_IO_POLICY,
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java
index 539676c..6cdcd79 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java
@@ -1422,7 +1422,6 @@ public class GridTaskWorker<T, R> extends GridWorker implements GridTimeoutObjec
forceLocDep,
ses.isFullSupport(),
internal,
- subjId,
affCacheIds,
affPartId,
mapTopVer,
@@ -1589,7 +1588,7 @@ public class GridTaskWorker<T, R> extends GridWorker implements GridTimeoutObjec
evt.taskNode(evtNode);
evt.jobId(jobId);
evt.type(evtType);
- evt.taskSubjectId(ses.subjectId());
+ evt.taskSubjectId(subjId);
evt.resultPolicy(plc);
ctx.event().record(evt);
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/GridJobSubjectIdSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/GridJobSubjectIdSelfTest.java
deleted file mode 100644
index 65fa402..0000000
--- a/modules/core/src/test/java/org/apache/ignite/internal/GridJobSubjectIdSelfTest.java
+++ /dev/null
@@ -1,210 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF 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.apache.ignite.internal;
-
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.UUID;
-import org.apache.ignite.Ignite;
-import org.apache.ignite.cluster.ClusterNode;
-import org.apache.ignite.compute.ComputeJob;
-import org.apache.ignite.compute.ComputeJobAdapter;
-import org.apache.ignite.compute.ComputeJobResult;
-import org.apache.ignite.compute.ComputeTaskAdapter;
-import org.apache.ignite.compute.ComputeTaskSession;
-import org.apache.ignite.configuration.IgniteConfiguration;
-import org.apache.ignite.events.Event;
-import org.apache.ignite.events.EventType;
-import org.apache.ignite.events.JobEvent;
-import org.apache.ignite.internal.util.typedef.F;
-import org.apache.ignite.lang.IgnitePredicate;
-import org.apache.ignite.resources.TaskSessionResource;
-import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
-import org.jetbrains.annotations.NotNull;
-import org.jetbrains.annotations.Nullable;
-import org.junit.Test;
-
-import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_SUBJ_ID;
-
-/**
- * Test job subject ID propagation.
- */
-public class GridJobSubjectIdSelfTest extends GridCommonAbstractTest {
- /** Job subject ID. */
- private static volatile UUID taskSubjId;
-
- /** Job subject ID. */
- private static volatile UUID jobSubjId;
-
- /** Event subject ID. */
- private static volatile UUID evtSubjId;
-
- /** First node. */
- private Ignite node1;
-
- /** Second node. */
- private Ignite node2;
-
- /** {@inheritDoc} */
- @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
- return super.getConfiguration(igniteInstanceName).setIncludeEventTypes(EventType.EVTS_ALL);
- }
-
- /** {@inheritDoc} */
- @Override protected void beforeTest() throws Exception {
- node1 = startGrid(1);
-
- node2 = startGrid(2);
- }
-
- /** {@inheritDoc} */
- @Override protected void afterTest() throws Exception {
- stopAllGrids();
-
- node1 = null;
-
- node2 = null;
-
- evtSubjId = null;
-
- taskSubjId = null;
-
- jobSubjId = null;
- }
-
- /**
- * Test job subject ID propagation.
- *
- * @throws Exception If failed.
- */
- @Test
- public void testJobSubjectId() throws Exception {
- node2.events().localListen(new IgnitePredicate<Event>() {
- @Override public boolean apply(Event evt) {
- JobEvent evt0 = (JobEvent)evt;
-
- assert evtSubjId == null;
-
- evtSubjId = evt0.taskSubjectId();
-
- return false;
- }
- }, EventType.EVT_JOB_STARTED);
-
- node1.compute().execute(new Task(node2.cluster().localNode().id()), null);
-
- assertEquals(taskSubjId, jobSubjId);
-
- assertEquals(taskSubjId, evtSubjId);
- }
-
- /**
- * Test job subject ID propagation in case if was changed.
- *
- * @throws Exception If failed.
- */
- @Test
- public void testModifiedSubjectId() throws Exception {
- node1.events().localListen(new IgnitePredicate<Event>() {
- @Override public boolean apply(Event evt) {
- JobEvent evt0 = (JobEvent)evt;
-
- assert evtSubjId == null;
-
- evtSubjId = evt0.taskSubjectId();
-
- return false;
- }
- }, EventType.EVT_JOB_STARTED);
-
- UUID uuid = new UUID(100, 100);
-
- ((IgniteEx) node1).context().task().setThreadContextIfNotNull(TC_SUBJ_ID, uuid);
-
- ((IgniteEx) node1).context().task().execute(new Task(node1.cluster().localNode().id()), null).get();
-
- assertEquals(uuid, jobSubjId);
-
- assertEquals(uuid, evtSubjId);
- }
-
- /**
- * Task class.
- */
- @SuppressWarnings("PublicInnerClass")
- public static class Task extends ComputeTaskAdapter<Object, Object> {
- /** Target node ID. */
- private UUID targetNodeId;
-
- /** Session. */
- @TaskSessionResource
- private ComputeTaskSession ses;
-
- /**
- * Constructor.
- *
- * @param targetNodeId Target node ID.
- */
- public Task(UUID targetNodeId) {
- this.targetNodeId = targetNodeId;
- }
-
- /** {@inheritDoc} */
- @NotNull @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid,
- @Nullable Object arg) {
- taskSubjId = ((GridTaskSessionInternal)ses).subjectId();
-
- ClusterNode node = null;
-
- for (ClusterNode subgridNode : subgrid) {
- if (F.eq(targetNodeId, subgridNode.id())) {
- node = subgridNode;
-
- break;
- }
- }
-
- assert node != null;
-
- return Collections.singletonMap(new Job(), node);
- }
-
- /** {@inheritDoc} */
- @Nullable @Override public Object reduce(List<ComputeJobResult> results) {
- return null;
- }
- }
-
- /**
- * Job class.
- */
- @SuppressWarnings("PublicInnerClass")
- public static class Job extends ComputeJobAdapter {
- /** Session. */
- @TaskSessionResource
- private ComputeTaskSession ses;
-
- /** {@inheritDoc} */
- @Nullable @Override public Object execute() {
- jobSubjId = ((GridTaskSessionInternal)ses).subjectId();
-
- return null;
- }
- }
-}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/managers/GridManagerStopSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/managers/GridManagerStopSelfTest.java
index 79ebf2a..4c30c84 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/managers/GridManagerStopSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/managers/GridManagerStopSelfTest.java
@@ -32,6 +32,7 @@ import org.apache.ignite.internal.managers.systemview.GridSystemViewManager;
import org.apache.ignite.internal.processors.metric.GridMetricManager;
import org.apache.ignite.internal.processors.pool.PoolProcessor;
import org.apache.ignite.internal.processors.resource.GridResourceProcessor;
+import org.apache.ignite.internal.processors.security.NoOpIgniteSecurityProcessor;
import org.apache.ignite.resources.LoggerResource;
import org.apache.ignite.spi.IgniteSpi;
import org.apache.ignite.spi.checkpoint.sharedfs.SharedFsCheckpointSpi;
@@ -73,6 +74,7 @@ public class GridManagerStopSelfTest extends GridCommonAbstractTest {
ctx.config().setMBeanServer(ManagementFactory.getPlatformMBeanServer());
ctx.add(new PoolProcessor(ctx));
+ ctx.add(new NoOpIgniteSecurityProcessor(ctx));
ctx.add(new GridResourceProcessor(ctx));
ctx.add(new GridSystemViewManager(ctx));
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/security/impl/TestCertificateSecurityProcessor.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/security/impl/TestCertificateSecurityProcessor.java
index a59f644..f4a2fa6 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/security/impl/TestCertificateSecurityProcessor.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/security/impl/TestCertificateSecurityProcessor.java
@@ -54,6 +54,9 @@ public class TestCertificateSecurityProcessor extends GridProcessorAdapter imple
/** Permissions. */
public static final Map<String, SecurityPermissionSet> PERMS = new ConcurrentHashMap<>();
+ /** Key - security subject ID, value - security contexts referred to this security subject. */
+ private final Map<UUID, SecurityContext> secCtxs = new ConcurrentHashMap<>();
+
/** Users security data. */
private final Collection<TestSecurityData> predefinedAuthData;
@@ -101,7 +104,7 @@ public class TestCertificateSecurityProcessor extends GridProcessorAdapter imple
if (!PERMS.containsKey(cn))
return null;
- return new TestSecurityContext(
+ SecurityContext res = new TestSecurityContext(
new TestSecuritySubject()
.setType(ctx.subjectType())
.setId(ctx.subjectId())
@@ -110,6 +113,10 @@ public class TestCertificateSecurityProcessor extends GridProcessorAdapter imple
.setPerms(PERMS.get(cn))
.setCerts(ctx.certificates())
);
+
+ secCtxs.put(res.subject().id(), res);
+
+ return res;
}
/** {@inheritDoc} */
@@ -123,6 +130,11 @@ public class TestCertificateSecurityProcessor extends GridProcessorAdapter imple
}
/** {@inheritDoc} */
+ @Override public SecurityContext securityContext(UUID subjId) {
+ return secCtxs.get(subjId);
+ }
+
+ /** {@inheritDoc} */
@Override public void authorize(String name, SecurityPermission perm, SecurityContext securityCtx)
throws SecurityException {
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/security/impl/TestSecurityProcessor.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/security/impl/TestSecurityProcessor.java
index 7e0464a..d7113bf 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/security/impl/TestSecurityProcessor.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/security/impl/TestSecurityProcessor.java
@@ -25,12 +25,14 @@ import java.util.Collections;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteNodeAttributes;
import org.apache.ignite.internal.processors.GridProcessorAdapter;
import org.apache.ignite.internal.processors.security.GridSecurityProcessor;
+import org.apache.ignite.internal.processors.security.IgniteSecurityProcessor.DeferredSecurityContext;
import org.apache.ignite.internal.processors.security.SecurityContext;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.plugin.security.AuthenticationContext;
@@ -85,7 +87,7 @@ public class TestSecurityProcessor extends GridProcessorAdapter implements GridS
if (!PERMS.containsKey(cred))
return null;
- return new TestSecurityContext(
+ SecurityContext res = new TestSecurityContext(
new TestSecuritySubject()
.setType(REMOTE_NODE)
.setId(node.id())
@@ -94,6 +96,10 @@ public class TestSecurityProcessor extends GridProcessorAdapter implements GridS
.setPerms(PERMS.get(cred))
.sandboxPermissions(SANDBOX_PERMS.get(cred))
);
+
+ SECURITY_CONTEXTS.put(res.subject().id(), res);
+
+ return res;
}
/** {@inheritDoc} */
@@ -131,12 +137,12 @@ public class TestSecurityProcessor extends GridProcessorAdapter implements GridS
/** {@inheritDoc} */
@Override public Collection<SecuritySubject> authenticatedSubjects() {
- return Collections.emptyList();
+ return SECURITY_CONTEXTS.values().stream().map(SecurityContext::subject).collect(Collectors.toList());
}
/** {@inheritDoc} */
@Override public SecuritySubject authenticatedSubject(UUID subjId) {
- return null;
+ return securityContext(subjId).subject();
}
/** {@inheritDoc} */
@@ -147,6 +153,9 @@ public class TestSecurityProcessor extends GridProcessorAdapter implements GridS
/** {@inheritDoc} */
@Override public void authorize(String name, SecurityPermission perm, SecurityContext securityCtx)
throws SecurityException {
+ if (securityCtx instanceof DeferredSecurityContext)
+ securityCtx = ((DeferredSecurityContext)securityCtx).delegate();
+
if (!((TestSecurityContext)securityCtx).operationAllowed(name, perm))
throw new SecurityException("Authorization failed [perm=" + perm +
", name=" + name +
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/future/GridFutureAdapterSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/util/future/GridFutureAdapterSelfTest.java
index e023dfe..e7e9e5d 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/util/future/GridFutureAdapterSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/util/future/GridFutureAdapterSelfTest.java
@@ -30,6 +30,7 @@ import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.cluster.ClusterGroupEmptyCheckedException;
import org.apache.ignite.internal.processors.closure.GridClosureProcessor;
import org.apache.ignite.internal.processors.pool.PoolProcessor;
+import org.apache.ignite.internal.processors.security.NoOpIgniteSecurityProcessor;
import org.apache.ignite.internal.util.typedef.CI1;
import org.apache.ignite.internal.util.typedef.CX1;
import org.apache.ignite.testframework.GridTestUtils;
@@ -179,6 +180,8 @@ public class GridFutureAdapterSelfTest extends GridCommonAbstractTest {
public void testListenNotify() throws Exception {
GridTestKernalContext ctx = new GridTestKernalContext(log);
+ ctx.add(new NoOpIgniteSecurityProcessor(ctx));
+
ctx.add(new PoolProcessor(ctx) {
final ExecutorService execSvc = Executors.newSingleThreadExecutor(new IgniteThreadFactory("testscope", "exec-svc"));
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TestReconnectProcessor.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TestReconnectProcessor.java
index 154d67c..6a8dfeb 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TestReconnectProcessor.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TestReconnectProcessor.java
@@ -71,6 +71,11 @@ public class TestReconnectProcessor extends GridProcessorAdapter implements Grid
}
/** {@inheritDoc} */
+ @Override public SecurityContext securityContext(UUID subjId) {
+ return new TestSecurityContext(new TestSecuritySubject(subjId));
+ }
+
+ /** {@inheritDoc} */
@Override public Collection<SecuritySubject> authenticatedSubjects() throws IgniteCheckedException {
return null;
}
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteClusterProcessProxy.java b/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteClusterProcessProxy.java
index 21e3773..30b9830 100644
--- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteClusterProcessProxy.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteClusterProcessProxy.java
@@ -33,7 +33,6 @@ import org.apache.ignite.cluster.ClusterMetrics;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.cluster.ClusterStartNodeResult;
import org.apache.ignite.cluster.ClusterState;
-import org.apache.ignite.internal.cluster.ClusterGroupEx;
import org.apache.ignite.internal.cluster.IgniteClusterEx;
import org.apache.ignite.internal.processors.cluster.baseline.autoadjust.BaselineAutoAdjustStatus;
import org.apache.ignite.lang.IgniteCallable;
@@ -63,11 +62,6 @@ public class IgniteClusterProcessProxy implements IgniteClusterEx {
}
/** {@inheritDoc} */
- @Override public ClusterGroupEx forSubjectId(UUID subjId) {
- throw new UnsupportedOperationException("Operation is not supported yet.");
- }
-
- /** {@inheritDoc} */
@Override public ClusterGroup forCacheNodes(@Nullable String cacheName, boolean affNodes, boolean nearNodes,
boolean clientNodes) {
throw new UnsupportedOperationException("Operation is not supported yet.");
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteComputeGridTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteComputeGridTestSuite.java
index d427f90..f63dfd5 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteComputeGridTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteComputeGridTestSuite.java
@@ -44,7 +44,6 @@ import org.apache.ignite.internal.GridJobMasterLeaveAwareSelfTest;
import org.apache.ignite.internal.GridJobServicesAddNodeTest;
import org.apache.ignite.internal.GridJobStealingSelfTest;
import org.apache.ignite.internal.GridJobStealingZeroActiveJobsSelfTest;
-import org.apache.ignite.internal.GridJobSubjectIdSelfTest;
import org.apache.ignite.internal.GridMultipleJobsSelfTest;
import org.apache.ignite.internal.GridMultipleSpisSelfTest;
import org.apache.ignite.internal.GridMultipleVersionsDeploymentSelfTest;
@@ -117,7 +116,6 @@ import org.junit.runners.Suite;
GridJobContextSelfTest.class,
GridJobMasterLeaveAwareSelfTest.class,
GridJobStealingSelfTest.class,
- GridJobSubjectIdSelfTest.class,
GridMultithreadedJobStealingSelfTest.class,
GridAlwaysFailoverSpiFailSelfTest.class,
GridTaskInstanceExecutionSelfTest.class,