This is an automated email from the ASF dual-hosted git repository.
petrov-mg 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 0ad769ef66c IGNITE-28754 Replaced SecurityAwareCustomMessageWrapper
with OperationContextMessage (#13346)
0ad769ef66c is described below
commit 0ad769ef66cbed3a7e7661834e410d4205c66257
Author: Vladimir Steshin <[email protected]>
AuthorDate: Thu Jul 16 16:28:48 2026 +0300
IGNITE-28754 Replaced SecurityAwareCustomMessageWrapper with
OperationContextMessage (#13346)
---
.../ignite/internal/CoreMessagesProvider.java | 2 -
.../managers/discovery/GridDiscoveryManager.java | 75 ++----
.../SecurityAwareCustomMessageWrapper.java | 75 ------
.../apache/ignite/internal/util/IgniteUtils.java | 12 +-
.../main/resources/META-INF/classnames.properties | 1 -
.../NodeSecurityContextPropagationTest.java | 3 -
.../context/OperationContextAttributesTest.java | 241 ------------------
.../OperationContextSendAttributesTest.java | 278 +++++++++++++++++++++
.../ignite/testsuites/SecurityTestSuite.java | 2 +
.../spi/discovery/zk/ZookeeperDiscoverySpi.java | 2 +-
.../zk/internal/ZkDiscoveryCustomEventData.java | 2 +-
.../discovery/zk/internal/ZkMessageFactory.java | 1 +
.../ZkOperationContextAwareCustomMessage.java | 84 +++++++
.../zk/internal/ZookeeperDiscoveryImpl.java | 60 +++--
.../zk/ZookeeperDiscoverySpiTestSuite4.java | 2 +
.../ZkOperationContextSendAttributesTest.java} | 17 +-
16 files changed, 456 insertions(+), 401 deletions(-)
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java
b/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java
index 2580a267e32..50ba117039d 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java
@@ -34,7 +34,6 @@ import
org.apache.ignite.internal.managers.communication.SessionChannelMessage;
import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoBean;
import org.apache.ignite.internal.managers.deployment.GridDeploymentRequest;
import org.apache.ignite.internal.managers.deployment.GridDeploymentResponse;
-import
org.apache.ignite.internal.managers.discovery.SecurityAwareCustomMessageWrapper;
import
org.apache.ignite.internal.managers.encryption.ChangeCacheEncryptionRequest;
import org.apache.ignite.internal.managers.encryption.EncryptionDataBagItem;
import
org.apache.ignite.internal.managers.encryption.GenerateEncryptionKeyRequest;
@@ -443,7 +442,6 @@ public class CoreMessagesProvider extends
AbstractMarshallableMessageFactoryProv
withNoSchema(FullMessage.class);
withNoSchema(InitMessage.class);
withNoSchema(CacheStatisticsModeChangeMessage.class);
- withNoSchema(SecurityAwareCustomMessageWrapper.class);
withNoSchema(MetadataRemoveAcceptedMessage.class);
withNoSchema(MetadataRemoveProposedMessage.class);
withNoSchema(WalStateFinishMessage.class);
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 7411e501b3c..23cc502b177 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
@@ -88,7 +88,6 @@ import
org.apache.ignite.internal.processors.cluster.ChangeGlobalStateFinishMess
import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateMessage;
import org.apache.ignite.internal.processors.cluster.DiscoveryDataClusterState;
import
org.apache.ignite.internal.processors.cluster.IGridClusterStateProcessor;
-import org.apache.ignite.internal.processors.security.IgniteSecurity;
import org.apache.ignite.internal.processors.security.SecurityContext;
import org.apache.ignite.internal.processors.tracing.messages.SpanContainer;
import org.apache.ignite.internal.systemview.ClusterNodeViewWalker;
@@ -134,7 +133,6 @@ import
org.apache.ignite.spi.discovery.DiscoveryDataBag.JoiningNodeDiscoveryData
import org.apache.ignite.spi.discovery.DiscoveryMetricsProvider;
import org.apache.ignite.spi.discovery.DiscoveryNotification;
import org.apache.ignite.spi.discovery.DiscoverySpi;
-import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
import org.apache.ignite.spi.discovery.DiscoverySpiDataExchange;
import org.apache.ignite.spi.discovery.DiscoverySpiHistorySupport;
import org.apache.ignite.spi.discovery.DiscoverySpiListener;
@@ -558,9 +556,7 @@ public class GridDiscoveryManager extends
GridManagerAdapter<DiscoverySpi> {
@Override public IgniteFuture<?> onDiscovery(DiscoveryNotification
notification) {
GridFutureAdapter<?> notificationFut = new
GridFutureAdapter<>();
- discoMsgNotifier.submit(notificationFut,
ctx.security().enabled()
- ? new SecurityAwareNotificationTask(notification)
- : new NotificationTask(notification));
+ discoMsgNotifier.submit(notificationFut, new
NotificationTask(notification));
IgniteFuture<?> fut = new IgniteFutureImpl<>(notificationFut);
@@ -743,7 +739,7 @@ public class GridDiscoveryManager extends
GridManagerAdapter<DiscoverySpi> {
}
if (type == EVT_DISCOVERY_CUSTOM_EVT) {
- for (Class cls = customMsg.getClass(); cls != null; cls =
cls.getSuperclass()) {
+ for (Class<?> cls = customMsg.getClass(); cls != null; cls
= cls.getSuperclass()) {
List<CustomEventListener<DiscoveryCustomMessage>> list
= customEvtLsnrs.get(cls);
if (list != null) {
@@ -917,43 +913,8 @@ public class GridDiscoveryManager extends
GridManagerAdapter<DiscoverySpi> {
discoEvtHnd.awaitDisconnectEvent();
}
- /**
- * Extends {@link NotificationTask} to run in a security context
owned by the initiator of the
- * discovery event.
- */
- class SecurityAwareNotificationTask extends NotificationTask {
- /** */
- public SecurityAwareNotificationTask(DiscoveryNotification
notification) {
- super(notification);
- }
-
- /** */
- @Override public void run() {
- DiscoverySpiCustomMessage customMsg =
notification.customMessage();
-
- if (customMsg instanceof
SecurityAwareCustomMessageWrapper) {
- UUID secSubjId =
((SecurityAwareCustomMessageWrapper)customMsg).securitySubjectId();
-
- try (Scope ignored =
ctx.security().withContext(secSubjId)) {
- super.run();
- }
- }
- else {
- SecurityContext initiatorNodeSecCtx =
nodeSecurityContext(
- marshaller,
- U.resolveClassLoader(ctx.config()),
- notification.getNode()
- );
-
- try (Scope ignored =
ctx.security().withContext(initiatorNodeSecCtx)) {
- super.run();
- }
- }
- }
- }
-
/** Represents task to handle discovery notification
asynchronously. */
- class NotificationTask implements Runnable {
+ private class NotificationTask implements Runnable {
/** */
protected final DiscoveryNotification notification;
@@ -965,9 +926,31 @@ public class GridDiscoveryManager extends
GridManagerAdapter<DiscoverySpi> {
/** {@inheritDoc} */
@Override public void run() {
synchronized (discoEvtMux) {
- onDiscovery0(notification);
+ try (Scope ignored =
withRemoteSecurityContext(notification.getNode())) {
+ onDiscovery0(notification);
+ }
}
}
+
+ /** */
+ private Scope withRemoteSecurityContext(ClusterNode node) {
+ if (ctx.security().enabled()) {
+ if (ctx.security().isDefaultContext()) {
+ SecurityContext initiatorNodeSecCtx =
nodeSecurityContext(
+ marshaller,
+ U.resolveClassLoader(ctx.config()),
+ node
+ );
+
+ return
ctx.security().withContext(initiatorNodeSecCtx);
+ }
+
+ // Verify that the Security Context currently attached
to the thread is valid.
+ ctx.security().securityContext();
+ }
+
+ return Scope.NOOP_SCOPE;
+ }
}
});
@@ -2340,11 +2323,7 @@ public class GridDiscoveryManager extends
GridManagerAdapter<DiscoverySpi> {
*/
public void sendCustomEvent(DiscoveryCustomMessage msg) throws
IgniteCheckedException {
try {
- IgniteSecurity security = ctx.security();
-
- getSpi().sendCustomEvent(security.enabled()
- ? new SecurityAwareCustomMessageWrapper(msg,
security.securityContext().subject().id())
- : msg);
+ getSpi().sendCustomEvent(msg);
}
catch (IgniteClientDisconnectedException e) {
IgniteFuture<?> reconnectFut =
ctx.cluster().clientReconnectFuture();
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/SecurityAwareCustomMessageWrapper.java
b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/SecurityAwareCustomMessageWrapper.java
deleted file mode 100644
index e9d33b8433c..00000000000
---
a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/SecurityAwareCustomMessageWrapper.java
+++ /dev/null
@@ -1,75 +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.managers.discovery;
-
-import java.util.UUID;
-import org.apache.ignite.internal.Order;
-import org.apache.ignite.plugin.extensions.communication.MessageFactory;
-import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
-import org.jetbrains.annotations.Nullable;
-
-/** Custom message wrapper with ID of security subject that initiated the
current message. */
-public class SecurityAwareCustomMessageWrapper implements
DiscoverySpiCustomMessage {
- /** Security subject ID. */
- @Order(0)
- UUID secSubjId;
-
- /** Original message. */
- @Order(1)
- DiscoveryCustomMessage delegate;
-
- /** Default constructor for {@link MessageFactory}. */
- public SecurityAwareCustomMessageWrapper() {
- // No-op.
- }
-
- /** */
- public SecurityAwareCustomMessageWrapper(DiscoveryCustomMessage delegate,
UUID secSubjId) {
- this.delegate = delegate;
- this.secSubjId = secSubjId;
- }
-
- /** Gets security Subject ID. */
- public UUID securitySubjectId() {
- return secSubjId;
- }
-
- /** {@inheritDoc} */
- @Override public boolean isMutable() {
- return delegate().isMutable();
- }
-
- /** {@inheritDoc} */
- @Override public boolean stopProcess() {
- return delegate().stopProcess();
- }
-
- /**
- * @return Delegate.
- */
- public DiscoveryCustomMessage delegate() {
- return delegate;
- }
-
- /** {@inheritDoc} */
- @Override public @Nullable DiscoverySpiCustomMessage ackMessage() {
- DiscoveryCustomMessage ack =
(DiscoveryCustomMessage)delegate().ackMessage();
-
- return ack == null ? null : new SecurityAwareCustomMessageWrapper(ack,
secSubjId);
- }
-}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
index 1f722faffc8..d3770f81e54 100755
---
a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
@@ -181,7 +181,6 @@ import
org.apache.ignite.internal.managers.deployment.GridDeployment;
import org.apache.ignite.internal.managers.deployment.GridDeploymentInfo;
import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
import org.apache.ignite.internal.managers.discovery.GridDiscoveryManager;
-import
org.apache.ignite.internal.managers.discovery.SecurityAwareCustomMessageWrapper;
import org.apache.ignite.internal.mxbean.IgniteStandardMXBean;
import
org.apache.ignite.internal.processors.cache.CacheDefaultBinaryAffinityKeyMapper;
import org.apache.ignite.internal.processors.cache.CacheObjectContext;
@@ -7723,7 +7722,7 @@ public abstract class IgniteUtils extends CommonUtils {
/** */
public static <T extends IgniteDataTransferObject>
IgniteDataTransferObjectSerializer<T> loadSerializer(Class<T> cls) {
try {
- Class cls0 = IgniteUtils.class.getClassLoader()
+ Class<?> cls0 = IgniteUtils.class.getClassLoader()
.loadClass(cls.getPackage().getName() + "." +
cls.getSimpleName() + "Serializer");
return
(IgniteDataTransferObjectSerializer<T>)cls0.getDeclaredConstructor().newInstance();
@@ -7735,13 +7734,14 @@ public abstract class IgniteUtils extends CommonUtils {
}
/**
- * Unwraps messsage if it is wrapped by {@link
SecurityAwareCustomMessageWrapper}.
+ * Unwraps messsage as {@link DiscoveryCustomMessage}.
*
* @param msg Message.
*/
- public static DiscoveryCustomMessage
unwrapCustomMessage(DiscoverySpiCustomMessage msg) {
- return msg instanceof SecurityAwareCustomMessageWrapper ?
- ((SecurityAwareCustomMessageWrapper)msg).delegate() :
(DiscoveryCustomMessage)msg;
+ public static DiscoveryCustomMessage unwrapCustomMessage(@Nullable
DiscoverySpiCustomMessage msg) {
+ assert msg == null || msg instanceof DiscoveryCustomMessage;
+
+ return (DiscoveryCustomMessage)msg;
}
/**
diff --git a/modules/core/src/main/resources/META-INF/classnames.properties
b/modules/core/src/main/resources/META-INF/classnames.properties
index 91b37c246ef..04ceccd84b4 100644
--- a/modules/core/src/main/resources/META-INF/classnames.properties
+++ b/modules/core/src/main/resources/META-INF/classnames.properties
@@ -725,7 +725,6 @@
org.apache.ignite.internal.managers.discovery.GridDiscoveryManager$1
org.apache.ignite.internal.managers.discovery.GridDiscoveryManager$3$1
org.apache.ignite.internal.managers.discovery.GridDiscoveryManager$6
org.apache.ignite.internal.managers.discovery.GridDiscoveryManager$7
-org.apache.ignite.internal.managers.discovery.SecurityAwareCustomMessageWrapper
org.apache.ignite.internal.managers.encryption.CacheGroupEncryptionKeys$TrackedWalSegment
org.apache.ignite.internal.managers.encryption.CacheGroupPageScanner$1
org.apache.ignite.internal.managers.encryption.CacheGroupPageScanner$2
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/security/NodeSecurityContextPropagationTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/security/NodeSecurityContextPropagationTest.java
index c0a4a503c71..a579c68ec8d 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/processors/security/NodeSecurityContextPropagationTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/security/NodeSecurityContextPropagationTest.java
@@ -30,7 +30,6 @@ import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.failure.StopNodeOrHaltFailureHandler;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.events.DiscoveryCustomEvent;
-import
org.apache.ignite.internal.managers.discovery.SecurityAwareCustomMessageWrapper;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.spi.MessagesPluginProvider;
import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
@@ -160,8 +159,6 @@ public class NodeSecurityContextPropagationTest extends
GridCommonAbstractTest {
if (msg instanceof TcpDiscoveryCustomEventMessage) {
DiscoverySpiCustomMessage customMsg =
((TcpDiscoveryCustomEventMessage)msg).message();
- assert customMsg instanceof SecurityAwareCustomMessageWrapper;
-
unwrappedMsg = U.unwrapCustomMessage(customMsg);
}
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/thread/context/OperationContextAttributesTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/thread/context/OperationContextAttributesTest.java
index 8255437422b..b5ecbefcd8d 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/thread/context/OperationContextAttributesTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/thread/context/OperationContextAttributesTest.java
@@ -17,17 +17,13 @@
package org.apache.ignite.internal.thread.context;
-import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
-import java.util.Set;
-import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Delayed;
import java.util.concurrent.ExecutorService;
@@ -39,20 +35,9 @@ import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
-import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteException;
-import org.apache.ignite.cluster.ClusterNode;
-import org.apache.ignite.configuration.IgniteConfiguration;
-import org.apache.ignite.internal.GridKernalContext;
-import org.apache.ignite.internal.GridTopic;
-import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.managers.communication.GridIoPolicy;
-import org.apache.ignite.internal.managers.communication.GridMessageListener;
-import org.apache.ignite.internal.managers.communication.IgniteIoTestMessage;
-import org.apache.ignite.internal.managers.discovery.CustomEventListener;
-import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
-import org.apache.ignite.internal.processors.cache.DynamicCacheChangeBatch;
import org.apache.ignite.internal.processors.timeout.GridTimeoutObject;
import org.apache.ignite.internal.processors.timeout.GridTimeoutProcessor;
import
org.apache.ignite.internal.thread.context.concurrent.IgniteCompletableFuture;
@@ -62,10 +47,7 @@ import
org.apache.ignite.internal.thread.pool.IgniteScheduledThreadPoolExecutor;
import org.apache.ignite.internal.thread.pool.IgniteStripedExecutor;
import org.apache.ignite.internal.thread.pool.IgniteStripedThreadPoolExecutor;
import org.apache.ignite.internal.thread.pool.IgniteThreadPoolExecutor;
-import org.apache.ignite.internal.util.GridByteArrayList;
-import org.apache.ignite.internal.util.GridIntList;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
-import org.apache.ignite.internal.util.typedef.G;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.internal.util.worker.queue.IgniteAsyncObjectHandler;
import org.apache.ignite.internal.util.worker.queue.IgniteDelayedObjectHandler;
@@ -74,21 +56,14 @@ import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgniteOutClosure;
import org.apache.ignite.lang.IgniteRunnable;
import org.apache.ignite.lang.IgniteUuid;
-import org.apache.ignite.plugin.AbstractTestPluginProvider;
-import org.apache.ignite.plugin.PluginContext;
-import org.apache.ignite.plugin.PluginProvider;
-import org.apache.ignite.spi.discovery.tcp.messages.InetSocketAddressMessage;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.apache.ignite.thread.IgniteThread;
import org.junit.Test;
import org.springframework.lang.NonNull;
-import org.springframework.lang.Nullable;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
-import static org.apache.ignite.testframework.GridTestUtils.assertThrows;
import static
org.apache.ignite.testframework.GridTestUtils.assertThrowsAnyCause;
import static
org.apache.ignite.testframework.GridTestUtils.assertThrowsWithCause;
-import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
/** */
public class OperationContextAttributesTest extends GridCommonAbstractTest {
@@ -110,9 +85,6 @@ public class OperationContextAttributesTest extends
GridCommonAbstractTest {
/** */
private int beforeTestReservedAttrIds;
- /** */
- private @Nullable PluginProvider pluginProvider;
-
/** {@inheritDoc} */
@Override protected void beforeTest() throws Exception {
super.beforeTest();
@@ -135,16 +107,6 @@ public class OperationContextAttributesTest extends
GridCommonAbstractTest {
OperationContextAttribute.ID_GEN.set(beforeTestReservedAttrIds);
}
- /** {@inheritDoc} */
- @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
- IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
-
- if (pluginProvider != null)
- cfg.setPluginProviders(pluginProvider);
-
- return cfg;
- }
-
/** */
@Test
public void testNotAttachedAttribute() {
@@ -848,209 +810,6 @@ public class OperationContextAttributesTest extends
GridCommonAbstractTest {
}
}
- /** */
- @Test
- public void testSendAttributesByDiscovery() throws Exception {
- doTestOperationContextAttributesPropagation(true);
- }
-
- /** */
- @Test
- public void testSendAttributesByCommunication() throws Exception {
- doTestOperationContextAttributesPropagation(false);
- }
-
- /** */
- private void doTestOperationContextAttributesPropagation(boolean
discovery) throws Exception {
- OperationContextAttribute<InetSocketAddressMessage> dAttr1 =
- OperationContextAttribute.newInstance(new
InetSocketAddressMessage(InetAddress.getLoopbackAddress(), 80));
-
- OperationContextAttribute<GridIntList> dAttr2 =
OperationContextAttribute.newInstance(new GridIntList(new int[]{1, 1, 1}));
-
- OperationContextAttribute<GridByteArrayList> otherTestAttr =
OperationContextAttribute.newInstance(new GridByteArrayList());
-
- pluginProvider = new AbstractTestPluginProvider() {
- @Override public String name() {
- return "TestDistributedOperationContextAttributesRegistrator";
- }
-
- @Override public void start(PluginContext ctx) {
- GridKernalContext kctx = ((IgniteEx)ctx.grid()).context();
-
- int dAttr1Id = OperationContextDispatcher.MAX_ATTRS_CNT - 2;
- int dAttr2Id = OperationContextDispatcher.MAX_ATTRS_CNT - 1;
-
-
kctx.operationContextDispatcher().registerDistributedAttribute(dAttr1Id,
dAttr1);
-
kctx.operationContextDispatcher().registerDistributedAttribute(dAttr2Id,
dAttr2);
-
- assertThrowsAnyCause(
- log,
- () -> {
-
kctx.operationContextDispatcher().registerDistributedAttribute(dAttr2Id,
otherTestAttr);
- return null;
-
- }, IgniteException.class,
- "Duplicated distributed attribute id"
- );
- }
- };
-
- // Local attribute 1.
- OperationContextAttribute.newInstance(1000);
-
- startGrids(2);
- startClientGrid(2);
-
- assertThrows(
- null,
- () ->
grid(0).context().operationContextDispatcher().registerDistributedAttribute(1,
null),
- IgniteException.class,
- "Initialization of distributed operation context attributes has
already finished"
- );
-
- // Local attribute 2.
- OperationContextAttribute.newInstance("locaAttr2");
-
- InetSocketAddressMessage valToSend1 = new
InetSocketAddressMessage(dAttr1.initialValue().address(), 443);
- GridIntList valToSend2 = new GridIntList(new int[]{2, 2, 2});
-
- if (discovery)
-
doTestOperationContextAttributesPropagationThroughDiscovery(dAttr1, valToSend1,
dAttr2, valToSend2);
- else
-
doTestOperationContextAttributesPropagationThroughCommunication(dAttr1,
valToSend1, dAttr2, valToSend2);
- }
-
- /** */
- private void doTestOperationContextAttributesPropagationThroughDiscovery(
- OperationContextAttribute<InetSocketAddressMessage> dAttr1,
- InetSocketAddressMessage valToSend1,
- OperationContextAttribute<GridIntList> dAttr2,
- GridIntList valToSend2
- ) throws Exception {
- Set<Integer> checkedNodes = ConcurrentHashMap.newKeySet();
-
- for (int i = 0; i < G.allGrids().size(); ++i) {
- int i0 = i;
-
- grid(i).context().discovery().setCustomEventListener(
- DynamicCacheChangeBatch.class, new CustomEventListener<>() {
- @Override public void
onCustomEvent(AffinityTopologyVersion topVer, ClusterNode snd,
- DynamicCacheChangeBatch msg) {
-
- InetSocketAddressMessage receivedVal1 =
OperationContext.get(dAttr1);
- GridIntList receivedVal2 =
OperationContext.get(dAttr2);
-
- assertTrue(receivedVal1 != null && valToSend1.port()
== receivedVal1.port());
- assertTrue(receivedVal1 != null &&
valToSend1.address().equals(receivedVal1.address()));
-
- assertEquals(valToSend2, receivedVal2);
-
- checkedNodes.add(i0);
- }
- });
- }
-
- // Send from the coordinator.
- try (Scope ignored = OperationContext.set(dAttr1, valToSend1, dAttr2,
valToSend2)) {
- grid(0).createCache(defaultCacheConfiguration());
- }
-
- assertTrue(waitForCondition(() -> checkedNodes.size() == 3,
getTestTimeout(), 50));
- checkedNodes.clear();
-
- // Send from a server.
- try (Scope ignored = OperationContext.set(dAttr1, valToSend1, dAttr2,
valToSend2)) {
- grid(1).destroyCache(DEFAULT_CACHE_NAME);
- }
-
- assertTrue(waitForCondition(() -> checkedNodes.size() == 3,
getTestTimeout(), 50));
- checkedNodes.clear();
-
- // Send from a client.
- try (Scope ignored = OperationContext.set(dAttr1, valToSend1, dAttr2,
valToSend2)) {
- grid(2).createCache(defaultCacheConfiguration());
- }
-
- assertTrue(waitForCondition(() -> checkedNodes.size() == 3,
getTestTimeout(), 50));
- checkedNodes.clear();
- }
-
- /** */
- private void
doTestOperationContextAttributesPropagationThroughCommunication(
- OperationContextAttribute<InetSocketAddressMessage> dAttr1,
- InetSocketAddressMessage valToSend1,
- OperationContextAttribute<GridIntList> dAttr2,
- GridIntList valToSend2
- ) throws Exception {
- // Coordinator -> Server, Coordinator -> Client, Server -> Client,
Client -> Server, etc.
- for (int fromIdx = 0; fromIdx < 3; ++fromIdx) {
- for (int toIdx = 0; toIdx < 3; ++toIdx) {
- if (fromIdx == toIdx)
- continue;
-
- // One value.
- try (Scope ignored = OperationContext.set(dAttr1, valToSend1))
{
- checkOperationContextCommunicationTransmission(fromIdx,
toIdx, dAttr1, null);
- }
-
- // A couple of values.
- try (Scope ignored = OperationContext.set(dAttr1, valToSend1,
dAttr2, valToSend2)) {
- checkOperationContextCommunicationTransmission(fromIdx,
toIdx, dAttr1, dAttr2);
- }
- }
- }
- }
-
- /** */
- private void checkOperationContextCommunicationTransmission(
- int gridFromIdx,
- int gridToIdx,
- OperationContextAttribute<InetSocketAddressMessage> attr1,
- @Nullable OperationContextAttribute<GridIntList> attr2
- ) throws Exception {
- IgniteEx from = grid(gridFromIdx);
- IgniteEx to = grid(gridToIdx);
-
- CountDownLatch rcvLatch = new CountDownLatch(2);
-
- InetSocketAddressMessage expVal1 = OperationContext.get(attr1);
- GridIntList expVal2 = attr2 == null ? null :
OperationContext.get(attr2);
-
- GridMessageListener lsnr = new GridMessageListener() {
- @Override public void onMessage(UUID nodeId, Object msg, byte plc)
{
- if (msg instanceof IgniteIoTestMessage &&
((IgniteIoTestMessage)msg).request()) {
- InetSocketAddressMessage receivedVal1 =
OperationContext.get(attr1);
- GridIntList receivedVal2 = attr2 == null ? null :
OperationContext.get(attr2);
-
- assertTrue(receivedVal1 != null && expVal1.port() ==
receivedVal1.port());
- assertTrue(receivedVal1 != null &&
expVal1.address().equals(receivedVal1.address()));
-
- if (attr2 != null)
- assertEquals(expVal2, receivedVal2);
-
- rcvLatch.countDown();
- }
- }
- };
-
- to.context().io().addMessageListener(GridTopic.TOPIC_IO_TEST, lsnr);
-
- try {
- from.context().io().sendIoTest(node(from, to), null, false);
- from.context().io().sendIoTest(node(from, to), null, true);
-
- assertTrue(rcvLatch.await(getTestTimeout(), MILLISECONDS));
- }
- finally {
-
assertTrue(to.context().io().removeMessageListener(GridTopic.TOPIC_IO_TEST,
lsnr));
- }
- }
-
- /** Prevents {@link ClusterNode#isLocal()} to be negative. */
- private ClusterNode node(Ignite from, Ignite to) {
- return from.cluster().node(((IgniteEx)to).localNode().id());
- }
-
/** */
private void doContextAwareExecutorServiceTest(ExecutorService pool)
throws Exception {
CountDownLatch poolUnblockedLatch = blockPool(pool);
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/thread/context/OperationContextSendAttributesTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/thread/context/OperationContextSendAttributesTest.java
new file mode 100644
index 00000000000..68b638bf610
--- /dev/null
+++
b/modules/core/src/test/java/org/apache/ignite/internal/thread/context/OperationContextSendAttributesTest.java
@@ -0,0 +1,278 @@
+/*
+ * 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.thread.context;
+
+import java.net.InetAddress;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.GridTopic;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.managers.communication.GridMessageListener;
+import org.apache.ignite.internal.managers.communication.IgniteIoTestMessage;
+import org.apache.ignite.internal.managers.discovery.CustomEventListener;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.DynamicCacheChangeBatch;
+import org.apache.ignite.internal.util.GridByteArrayList;
+import org.apache.ignite.internal.util.GridIntList;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.plugin.AbstractTestPluginProvider;
+import org.apache.ignite.plugin.PluginContext;
+import org.apache.ignite.plugin.PluginProvider;
+import org.apache.ignite.spi.discovery.tcp.messages.InetSocketAddressMessage;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.Test;
+import org.springframework.lang.Nullable;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.apache.ignite.testframework.GridTestUtils.assertThrows;
+import static
org.apache.ignite.testframework.GridTestUtils.assertThrowsAnyCause;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+
+/** */
+public class OperationContextSendAttributesTest extends GridCommonAbstractTest
{
+ /** */
+ private PluginProvider pluginProvider;
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ stopAllGrids();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ assert pluginProvider != null;
+
+ cfg.setPluginProviders(pluginProvider);
+
+ return cfg;
+ }
+
+ /** */
+ @Test
+ public void testSendAttributesByDiscovery() throws Exception {
+ doTestOperationContextAttributesPropagation(true);
+ }
+
+ /** */
+ @Test
+ public void testSendAttributesByCommunication() throws Exception {
+ doTestOperationContextAttributesPropagation(false);
+ }
+
+ /** */
+ protected void doTestOperationContextAttributesPropagation(boolean
discovery) throws Exception {
+ OperationContextAttribute<InetSocketAddressMessage> dAttr1 =
+ OperationContextAttribute.newInstance(new
InetSocketAddressMessage(InetAddress.getLoopbackAddress(), 80));
+
+ OperationContextAttribute<GridIntList> dAttr2 =
OperationContextAttribute.newInstance(new GridIntList(1));
+
+ OperationContextAttribute<GridByteArrayList> otherTestAttr =
OperationContextAttribute.newInstance(new GridByteArrayList());
+
+ pluginProvider = new AbstractTestPluginProvider() {
+ @Override public String name() {
+ return "TestDistributedOperationContextAttributesRegistrator";
+ }
+
+ @Override public void start(PluginContext ctx) {
+ GridKernalContext kctx = ((IgniteEx)ctx.grid()).context();
+
+ int dAttr1Id = OperationContextDispatcher.MAX_ATTRS_CNT - 2;
+ int dAttr2Id = OperationContextDispatcher.MAX_ATTRS_CNT - 1;
+
+
kctx.operationContextDispatcher().registerDistributedAttribute(dAttr1Id,
dAttr1);
+
kctx.operationContextDispatcher().registerDistributedAttribute(dAttr2Id,
dAttr2);
+
+ assertThrowsAnyCause(
+ log,
+ () -> {
+
kctx.operationContextDispatcher().registerDistributedAttribute(dAttr2Id,
otherTestAttr);
+ return null;
+
+ }, IgniteException.class,
+ "Duplicated distributed attribute id"
+ );
+ }
+ };
+
+ // Local attribute 1.
+ OperationContextAttribute.newInstance(1000);
+
+ startGrids(2);
+ startClientGrid(2);
+
+ assertThrows(
+ null,
+ () ->
grid(0).context().operationContextDispatcher().registerDistributedAttribute(1,
null),
+ IgniteException.class,
+ "Initialization of distributed operation context attributes has
already finished"
+ );
+
+ // Local attribute 2.
+ OperationContextAttribute.newInstance("locaAttr2");
+
+ InetSocketAddressMessage valToSend1 = new
InetSocketAddressMessage(dAttr1.initialValue().address(), 443);
+ GridIntList valToSend2 = new GridIntList(2);
+
+ if (discovery)
+
doTestOperationContextAttributesPropagationThroughDiscovery(dAttr1, valToSend1,
dAttr2, valToSend2);
+ else
+
doTestOperationContextAttributesPropagationThroughCommunication(dAttr1,
valToSend1, dAttr2, valToSend2);
+ }
+
+ /** */
+ private void doTestOperationContextAttributesPropagationThroughDiscovery(
+ OperationContextAttribute<InetSocketAddressMessage> dAttr1,
+ InetSocketAddressMessage valToSend1,
+ OperationContextAttribute<GridIntList> dAttr2,
+ GridIntList valToSend2
+ ) throws Exception {
+ Set<Integer> checkedNodes = ConcurrentHashMap.newKeySet();
+
+ for (int i = 0; i < G.allGrids().size(); ++i) {
+ int i0 = i;
+
+ grid(i).context().discovery().setCustomEventListener(
+ DynamicCacheChangeBatch.class, new CustomEventListener<>() {
+ @Override public void
onCustomEvent(AffinityTopologyVersion topVer, ClusterNode snd,
+ DynamicCacheChangeBatch msg) {
+
+ InetSocketAddressMessage receivedVal1 =
OperationContext.get(dAttr1);
+ GridIntList receivedVal2 =
OperationContext.get(dAttr2);
+
+ assertTrue(receivedVal1 != null && valToSend1.port()
== receivedVal1.port());
+ assertTrue(receivedVal1 != null &&
valToSend1.address().equals(receivedVal1.address()));
+
+ assertEquals(valToSend2, receivedVal2);
+
+ checkedNodes.add(i0);
+ }
+ });
+ }
+
+ // Send from the coordinator.
+ try (Scope ignored = OperationContext.set(dAttr1, valToSend1, dAttr2,
valToSend2)) {
+ grid(0).createCache(defaultCacheConfiguration());
+ }
+
+ assertTrue(waitForCondition(() -> checkedNodes.size() == 3,
getTestTimeout(), 50));
+ checkedNodes.clear();
+
+ // Send from a server.
+ try (Scope ignored = OperationContext.set(dAttr1, valToSend1, dAttr2,
valToSend2)) {
+ grid(1).destroyCache(DEFAULT_CACHE_NAME);
+ }
+
+ assertTrue(waitForCondition(() -> checkedNodes.size() == 3,
getTestTimeout(), 50));
+ checkedNodes.clear();
+
+ // Send from a client.
+ try (Scope ignored = OperationContext.set(dAttr1, valToSend1, dAttr2,
valToSend2)) {
+ grid(2).createCache(defaultCacheConfiguration());
+ }
+
+ assertTrue(waitForCondition(() -> checkedNodes.size() == 3,
getTestTimeout(), 50));
+ checkedNodes.clear();
+ }
+
+ /** */
+ private void
doTestOperationContextAttributesPropagationThroughCommunication(
+ OperationContextAttribute<InetSocketAddressMessage> dAttr1,
+ InetSocketAddressMessage valToSend1,
+ OperationContextAttribute<GridIntList> dAttr2,
+ GridIntList valToSend2
+ ) throws Exception {
+ // Coordinator -> Server, Coordinator -> Client, Server -> Client,
Client -> Server, etc.
+ for (int fromIdx = 0; fromIdx < 3; ++fromIdx) {
+ for (int toIdx = 0; toIdx < 3; ++toIdx) {
+ if (fromIdx == toIdx)
+ continue;
+
+ // One value.
+ try (Scope ignored = OperationContext.set(dAttr1, valToSend1))
{
+ checkOperationContextCommunicationTransmission(fromIdx,
toIdx, dAttr1, null);
+ }
+
+ // A couple of values.
+ try (Scope ignored = OperationContext.set(dAttr1, valToSend1,
dAttr2, valToSend2)) {
+ checkOperationContextCommunicationTransmission(fromIdx,
toIdx, dAttr1, dAttr2);
+ }
+ }
+ }
+ }
+
+ /** */
+ private void checkOperationContextCommunicationTransmission(
+ int gridFromIdx,
+ int gridToIdx,
+ OperationContextAttribute<InetSocketAddressMessage> attr1,
+ @Nullable OperationContextAttribute<GridIntList> attr2
+ ) throws Exception {
+ IgniteEx from = grid(gridFromIdx);
+ IgniteEx to = grid(gridToIdx);
+
+ CountDownLatch rcvLatch = new CountDownLatch(2);
+
+ InetSocketAddressMessage expVal1 = OperationContext.get(attr1);
+ GridIntList expVal2 = attr2 == null ? null :
OperationContext.get(attr2);
+
+ GridMessageListener lsnr = new GridMessageListener() {
+ @Override public void onMessage(UUID nodeId, Object msg, byte plc)
{
+ if (msg instanceof IgniteIoTestMessage &&
((IgniteIoTestMessage)msg).request()) {
+ InetSocketAddressMessage receivedVal1 =
OperationContext.get(attr1);
+ GridIntList receivedVal2 = attr2 == null ? null :
OperationContext.get(attr2);
+
+ assertTrue(receivedVal1 != null && expVal1.port() ==
receivedVal1.port());
+ assertTrue(receivedVal1 != null &&
expVal1.address().equals(receivedVal1.address()));
+
+ if (attr2 != null)
+ assertEquals(expVal2, receivedVal2);
+
+ rcvLatch.countDown();
+ }
+ }
+ };
+
+ to.context().io().addMessageListener(GridTopic.TOPIC_IO_TEST, lsnr);
+
+ try {
+ from.context().io().sendIoTest(node(from, to), null, false);
+ from.context().io().sendIoTest(node(from, to), null, true);
+
+ assertTrue(rcvLatch.await(getTestTimeout(), MILLISECONDS));
+ }
+ finally {
+
assertTrue(to.context().io().removeMessageListener(GridTopic.TOPIC_IO_TEST,
lsnr));
+ }
+ }
+
+ /** Prevents {@link ClusterNode#isLocal()} to be negative. */
+ private ClusterNode node(Ignite from, Ignite to) {
+ return from.cluster().node(((IgniteEx)to).localNode().id());
+ }
+}
diff --git
a/modules/core/src/test/java/org/apache/ignite/testsuites/SecurityTestSuite.java
b/modules/core/src/test/java/org/apache/ignite/testsuites/SecurityTestSuite.java
index ba4e875162c..f7c0c688926 100644
---
a/modules/core/src/test/java/org/apache/ignite/testsuites/SecurityTestSuite.java
+++
b/modules/core/src/test/java/org/apache/ignite/testsuites/SecurityTestSuite.java
@@ -74,6 +74,7 @@ import
org.apache.ignite.internal.processors.security.service.ServiceAuthorizati
import
org.apache.ignite.internal.processors.security.service.ServiceStaticConfigTest;
import
org.apache.ignite.internal.processors.security.snapshot.SnapshotPermissionCheckTest;
import
org.apache.ignite.internal.thread.context.OperationContextAttributesTest;
+import
org.apache.ignite.internal.thread.context.OperationContextSendAttributesTest;
import org.apache.ignite.ssl.MultipleSSLContextsTest;
import org.apache.ignite.tools.junit.JUnitTeamcityReporter;
import org.junit.BeforeClass;
@@ -147,6 +148,7 @@ import org.junit.runners.Suite;
SecurityContextInternalFuturePropagationTest.class,
NodeConnectionCertificateCapturingTest.class,
OperationContextAttributesTest.class,
+ OperationContextSendAttributesTest.class,
})
public class SecurityTestSuite {
/** */
diff --git
a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/ZookeeperDiscoverySpi.java
b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/ZookeeperDiscoverySpi.java
index 1d0f2c68111..8f9b3008583 100644
---
a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/ZookeeperDiscoverySpi.java
+++
b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/ZookeeperDiscoverySpi.java
@@ -407,7 +407,7 @@ public class ZookeeperDiscoverySpi extends IgniteSpiAdapter
implements IgniteDis
/** {@inheritDoc} */
@Override public void sendCustomEvent(DiscoverySpiCustomMessage msg) {
- impl.sendCustomMessage(msg);
+ impl.sendCustomEvent(msg);
}
/** {@inheritDoc} */
diff --git
a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkDiscoveryCustomEventData.java
b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkDiscoveryCustomEventData.java
index e606f384fcf..9513108f836 100644
---
a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkDiscoveryCustomEventData.java
+++
b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkDiscoveryCustomEventData.java
@@ -40,7 +40,7 @@ class ZkDiscoveryCustomEventData extends ZkDiscoveryEventData
{
/** Message (can be marshalled as part of ZkDiscoveryCustomEventData or
stored in separate znode. */
byte[] msgBytes;
- /** Unmarshalled message. */
+ /** Unmarshalled custom message holder. Can be wrapped with {@link
ZkOperationContextAwareCustomMessage}. */
transient DiscoverySpiCustomMessage resolvedMsg;
/**
diff --git
a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkMessageFactory.java
b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkMessageFactory.java
index cf9f95a908f..e89f44f013b 100644
---
a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkMessageFactory.java
+++
b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkMessageFactory.java
@@ -29,5 +29,6 @@ public class ZkMessageFactory implements
MessageFactoryProvider {
factory.register(402, ZkForceNodeFailMessage::new, new
ZkForceNodeFailMessageSerializer());
factory.register(403, ZkNoServersMessage::new, new
ZkNoServersMessageSerializer());
factory.register(404, ZkDiscoDataBagWrapper::new, new
ZkDiscoDataBagWrapperSerializer());
+ factory.register(405, ZkOperationContextAwareCustomMessage::new, new
ZkOperationContextAwareCustomMessageSerializer());
}
}
diff --git
a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkOperationContextAwareCustomMessage.java
b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkOperationContextAwareCustomMessage.java
new file mode 100644
index 00000000000..4658c9aa75b
--- /dev/null
+++
b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkOperationContextAwareCustomMessage.java
@@ -0,0 +1,84 @@
+/*
+ * 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.spi.discovery.zk.internal;
+
+import org.apache.ignite.internal.OperationContextMessage;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.thread.context.OperationContext;
+import org.apache.ignite.internal.thread.context.OperationContextDispatcher;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
+import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import
org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
+import org.apache.ignite.spi.discovery.zk.ZookeeperDiscoverySpi;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * <p>A holder for effective attributes of distributed {@link
OperationContext}. Analogue of
+ * {@link TcpDiscoveryAbstractMessage#opCtxMsg} while we do not use a common
base class for all the Zookeeper messages.</p>
+ *
+ * <p>NOTE: The difference is also the limitation on message type. In {@link
TcpDiscoverySpi} we transfer distributed
+ * {@link OperationContext} with all the messages. In {@link
ZookeeperDiscoverySpi} with from-Ignite
+ * {@link DiscoverySpiCustomMessage} only.</p>
+ *
+ * @see OperationContextDispatcher
+ * @see ZookeeperDiscoverySpi#sendCustomEvent(DiscoverySpiCustomMessage)
+ */
+public class ZkOperationContextAwareCustomMessage implements
DiscoverySpiCustomMessage {
+ /** */
+ @Order(0)
+ DiscoverySpiCustomMessage delegate;
+
+ /** */
+ @Order(1)
+ OperationContextMessage opCtxMsg;
+
+ /** Default constructor for {@link MessageFactory}. */
+ public ZkOperationContextAwareCustomMessage() {
+ // No-op.
+ }
+
+ /**
+ * @param delegate Original message.
+ * @param opCtxMsg Distributed operation context message.
+ */
+ public ZkOperationContextAwareCustomMessage(DiscoverySpiCustomMessage
delegate, OperationContextMessage opCtxMsg) {
+ assert delegate != null;
+ assert opCtxMsg != null;
+ assert !(delegate instanceof ZkOperationContextAwareCustomMessage);
+
+ this.delegate = delegate;
+ this.opCtxMsg = opCtxMsg;
+ }
+
+ /** {@inheritDoc} */
+ @Override public @Nullable DiscoverySpiCustomMessage ackMessage() {
+ DiscoverySpiCustomMessage ack = delegate.ackMessage();
+ return ack == null ? null : new
ZkOperationContextAwareCustomMessage(ack, opCtxMsg);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean isMutable() {
+ return delegate.isMutable();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean stopProcess() {
+ return delegate.stopProcess();
+ }
+}
diff --git
a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoveryImpl.java
b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoveryImpl.java
index 8043589aba8..a4a657d0877 100644
---
a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoveryImpl.java
+++
b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoveryImpl.java
@@ -58,13 +58,17 @@ import
org.apache.ignite.configuration.CommunicationFailureResolver;
import org.apache.ignite.events.EventType;
import org.apache.ignite.events.NodeValidationFailedEvent;
import org.apache.ignite.internal.IgniteClientDisconnectedCheckedException;
+import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.IgnitionEx;
+import org.apache.ignite.internal.OperationContextMessage;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.events.DiscoveryCustomEvent;
import org.apache.ignite.internal.processors.security.SecurityContext;
+import org.apache.ignite.internal.thread.context.OperationContextDispatcher;
+import org.apache.ignite.internal.thread.context.Scope;
import org.apache.ignite.internal.thread.pool.IgniteThreadPoolExecutor;
import org.apache.ignite.internal.util.GridLongList;
import org.apache.ignite.internal.util.GridSpinBusyLock;
@@ -224,6 +228,9 @@ public class ZookeeperDiscoveryImpl {
/** */
private final DiscoveryMessageParser msgParser;
+ /** */
+ private final OperationContextDispatcher opCtxDispatcher;
+
/**
* @param spi Discovery SPI.
* @param igniteInstanceName Instance name.
@@ -274,6 +281,8 @@ public class ZookeeperDiscoveryImpl {
this.stats = stats;
msgParser = new DiscoveryMessageParser(msgFactory);
+
+ opCtxDispatcher =
((IgniteEx)spi.ignite()).context().operationContextDispatcher();
}
/**
@@ -658,10 +667,20 @@ public class ZookeeperDiscoveryImpl {
}
}
+ /** */
+ public void sendCustomEvent(DiscoverySpiCustomMessage msg) {
+ OperationContextMessage opCtx =
opCtxDispatcher.collectDistributedAttributes();
+
+ if (opCtx != null)
+ sendCustomMessage(new ZkOperationContextAwareCustomMessage(msg,
opCtx));
+ else
+ sendCustomMessage(msg);
+ }
+
/**
* @param msg Message.
*/
- public void sendCustomMessage(DiscoverySpiCustomMessage msg) {
+ void sendCustomMessage(DiscoverySpiCustomMessage msg) {
assert msg != null;
List<ClusterNode> nodes = rtState.top.topologySnapshot();
@@ -3503,10 +3522,19 @@ public class ZookeeperDiscoveryImpl {
}
/**
+ * Notifies the {@link DiscoverySpiListener} listener of a custom event.
Is aware of {@link ZkOperationContextAwareCustomMessage}.
+ *
* @param evtData Event data.
- * @param msg Custom message.
+ * @param msg Custom message to process. Can be a {@link
ZkOperationContextAwareCustomMessage}.
*/
- private void notifyCustomEvent(final ZkDiscoveryCustomEventData evtData,
final DiscoverySpiCustomMessage msg) {
+ private void notifyCustomEvent(final ZkDiscoveryCustomEventData evtData,
DiscoverySpiCustomMessage msg) {
+ OperationContextMessage opCtxMsg = null;
+
+ if (msg instanceof ZkOperationContextAwareCustomMessage) {
+ opCtxMsg = ((ZkOperationContextAwareCustomMessage)msg).opCtxMsg;
+ msg = ((ZkOperationContextAwareCustomMessage)msg).delegate;
+ }
+
assert !(msg instanceof ZkInternalMessage) : msg;
if (log.isDebugEnabled())
@@ -3518,17 +3546,21 @@ public class ZookeeperDiscoveryImpl {
final List<ClusterNode> topSnapshot = rtState.top.topologySnapshot();
- IgniteFuture<?> fut = lsnr.onDiscovery(
- new DiscoveryNotification(
- DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT,
- evtData.topologyVersion(),
- sndNode,
- topSnapshot,
- Collections.emptyNavigableMap(),
- msg,
- null
- )
- );
+ IgniteFuture<?> fut;
+
+ try (Scope ignored =
opCtxDispatcher.restoreDistributedAttributes(opCtxMsg)) {
+ fut = lsnr.onDiscovery(
+ new DiscoveryNotification(
+ DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT,
+ evtData.topologyVersion(),
+ sndNode,
+ topSnapshot,
+ Collections.emptyNavigableMap(),
+ msg,
+ null
+ )
+ );
+ }
if (msg != null && msg.isMutable())
fut.get();
diff --git
a/modules/zookeeper/src/test/java/org/apache/ignite/spi/discovery/zk/ZookeeperDiscoverySpiTestSuite4.java
b/modules/zookeeper/src/test/java/org/apache/ignite/spi/discovery/zk/ZookeeperDiscoverySpiTestSuite4.java
index e364bc12182..d468897a0e3 100644
---
a/modules/zookeeper/src/test/java/org/apache/ignite/spi/discovery/zk/ZookeeperDiscoverySpiTestSuite4.java
+++
b/modules/zookeeper/src/test/java/org/apache/ignite/spi/discovery/zk/ZookeeperDiscoverySpiTestSuite4.java
@@ -30,6 +30,7 @@ import
org.apache.ignite.internal.processors.metastorage.DistributedMetaStorageT
import
org.apache.ignite.internal.processors.security.cluster.ActivationOnJoinWithoutPermissionsWithPersistenceTest;
import
org.apache.ignite.internal.processors.security.cluster.NodeJoinPermissionsTest;
import org.apache.ignite.spi.discovery.DiscoverySpiDataExchangeTest;
+import
org.apache.ignite.spi.discovery.zk.internal.ZkOperationContextSendAttributesTest;
import org.junit.BeforeClass;
import org.junit.runner.RunWith;
import org.junit.runners.Suite;
@@ -50,6 +51,7 @@ import org.junit.runners.Suite;
DistributedMetaStoragePersistentTest.class,
IgniteNodeValidationFailedEventTest.class,
DiscoverySpiDataExchangeTest.class,
+ ZkOperationContextSendAttributesTest.class,
CacheCreateDestroyEventSecurityContextTest.class,
NodeJoinPermissionsTest.class,
ActivationOnJoinWithoutPermissionsWithPersistenceTest.class,
diff --git
a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkMessageFactory.java
b/modules/zookeeper/src/test/java/org/apache/ignite/spi/discovery/zk/internal/ZkOperationContextSendAttributesTest.java
similarity index 51%
copy from
modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkMessageFactory.java
copy to
modules/zookeeper/src/test/java/org/apache/ignite/spi/discovery/zk/internal/ZkOperationContextSendAttributesTest.java
index cf9f95a908f..c3d33bf5815 100644
---
a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkMessageFactory.java
+++
b/modules/zookeeper/src/test/java/org/apache/ignite/spi/discovery/zk/internal/ZkOperationContextSendAttributesTest.java
@@ -17,17 +17,16 @@
package org.apache.ignite.spi.discovery.zk.internal;
-import org.apache.ignite.plugin.extensions.communication.MessageFactory;
-import
org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
+import
org.apache.ignite.internal.thread.context.OperationContextSendAttributesTest;
+
+import static org.junit.Assume.assumeTrue;
/** */
-public class ZkMessageFactory implements MessageFactoryProvider {
+public class ZkOperationContextSendAttributesTest extends
OperationContextSendAttributesTest {
/** {@inheritDoc} */
- @Override public void registerAll(MessageFactory factory) {
- factory.register(400, ZkCommunicationErrorResolveFinishMessage::new,
new ZkCommunicationErrorResolveFinishMessageSerializer());
- factory.register(401, ZkCommunicationErrorResolveStartMessage::new,
new ZkCommunicationErrorResolveStartMessageSerializer());
- factory.register(402, ZkForceNodeFailMessage::new, new
ZkForceNodeFailMessageSerializer());
- factory.register(403, ZkNoServersMessage::new, new
ZkNoServersMessageSerializer());
- factory.register(404, ZkDiscoDataBagWrapper::new, new
ZkDiscoDataBagWrapperSerializer());
+ @Override protected void
doTestOperationContextAttributesPropagation(boolean discovery) throws Exception
{
+ assumeTrue(discovery);
+
+ super.doTestOperationContextAttributesPropagation(true);
}
}
