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 2a3d28888aac8..7cf22d101c22a 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.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; @@ -436,7 +435,6 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh, C 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 7411e501b3c51..23cc502b17775 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.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.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 @@ private void updateClientNodes(UUID leftNodeId) { @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 @@ else if (customMsg instanceof ChangeGlobalStateMessage) { } 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> list = customEvtLsnrs.get(cls); if (list != null) { @@ -917,43 +913,8 @@ else if (type == EVT_CLIENT_NODE_RECONNECTED) { 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 NotificationTask(DiscoveryNotification notification) { /** {@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 GridFutureAdapter localJoinFuture() { */ 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 e9d33b8433cbf..0000000000000 --- 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 1f722faffc8de..d3770f81e541e 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.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 static void prepareAffinityField(BinaryObjectBuilder builder, CacheObject /** */ public static IgniteDataTransferObjectSerializer loadSerializer(Class cls) { try { - Class cls0 = IgniteUtils.class.getClassLoader() + Class cls0 = IgniteUtils.class.getClassLoader() .loadClass(cls.getPackage().getName() + "." + cls.getSimpleName() + "Serializer"); return (IgniteDataTransferObjectSerializer)cls0.getDeclaredConstructor().newInstance(); @@ -7735,13 +7734,14 @@ public static IgniteDataTransferObjectSeria } /** - * 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 431e7ee68b9b7..411e26a00c38c 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 6e33bed666079..701a003b3881d 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 @@ -34,7 +34,6 @@ 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.DiscoverySpi; @@ -186,8 +185,6 @@ private boolean anyReceivedMessageMatch(IgniteEx ignite, Predicate predi 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 8255437422b9e..b5ecbefcd8d3a 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.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.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.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 void testContextAwareDelayQueue() throws Exception { } } - /** */ - @Test - public void testSendAttributesByDiscovery() throws Exception { - doTestOperationContextAttributesPropagation(true); - } - - /** */ - @Test - public void testSendAttributesByCommunication() throws Exception { - doTestOperationContextAttributesPropagation(false); - } - - /** */ - private void doTestOperationContextAttributesPropagation(boolean discovery) throws Exception { - OperationContextAttribute dAttr1 = - OperationContextAttribute.newInstance(new InetSocketAddressMessage(InetAddress.getLoopbackAddress(), 80)); - - OperationContextAttribute dAttr2 = OperationContextAttribute.newInstance(new GridIntList(new int[]{1, 1, 1})); - - OperationContextAttribute 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 dAttr1, - InetSocketAddressMessage valToSend1, - OperationContextAttribute dAttr2, - GridIntList valToSend2 - ) throws Exception { - Set 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 dAttr1, - InetSocketAddressMessage valToSend1, - OperationContextAttribute 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 attr1, - @Nullable OperationContextAttribute 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 0000000000000..68b638bf610f0 --- /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 dAttr1 = + OperationContextAttribute.newInstance(new InetSocketAddressMessage(InetAddress.getLoopbackAddress(), 80)); + + OperationContextAttribute dAttr2 = OperationContextAttribute.newInstance(new GridIntList(1)); + + OperationContextAttribute 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 dAttr1, + InetSocketAddressMessage valToSend1, + OperationContextAttribute dAttr2, + GridIntList valToSend2 + ) throws Exception { + Set 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 dAttr1, + InetSocketAddressMessage valToSend1, + OperationContextAttribute 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 attr1, + @Nullable OperationContextAttribute 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 ba4e875162cad..f7c0c688926bb 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.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 @@ 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 1d0f2c6811165..8f9b30085838b 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 DiscoverySpiNodeAuthenticator getAuthenticator() { /** {@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 e606f384fcf0f..9513108f8361b 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 cf9f95a908f7e..e89f44f013bb7 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 0000000000000..4658c9aa75b85 --- /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; + +/** + *

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.

+ * + *

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.

+ * + * @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 f69e9f6d10658..2c5ebe4459b6a 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 @@ -56,13 +56,17 @@ 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; @@ -222,6 +226,9 @@ public class ZookeeperDiscoveryImpl { /** */ private final DiscoveryMessageParser msgParser; + /** */ + private final OperationContextDispatcher opCtxDispatcher; + /** * @param spi Discovery SPI. * @param igniteInstanceName Instance name. @@ -272,6 +279,8 @@ public ZookeeperDiscoveryImpl( this.stats = stats; msgParser = new DiscoveryMessageParser(msgFactory); + + opCtxDispatcher = ((IgniteEx)spi.ignite()).context().operationContextDispatcher(); } /** @@ -656,10 +665,20 @@ public boolean knownNode(UUID nodeId) { } } + /** */ + 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 nodes = rtState.top.topologySnapshot(); @@ -3501,10 +3520,19 @@ public void simulateNodeFailure() { } /** + * 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()) @@ -3516,17 +3544,21 @@ private void notifyCustomEvent(final ZkDiscoveryCustomEventData evtData, final D final List 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 e364bc1218263..d468897a0e3f3 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.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 @@ DistributedMetaStoragePersistentTest.class, IgniteNodeValidationFailedEventTest.class, DiscoverySpiDataExchangeTest.class, + ZkOperationContextSendAttributesTest.class, CacheCreateDestroyEventSecurityContextTest.class, NodeJoinPermissionsTest.class, ActivationOnJoinWithoutPermissionsWithPersistenceTest.class, diff --git a/modules/zookeeper/src/test/java/org/apache/ignite/spi/discovery/zk/internal/ZkOperationContextSendAttributesTest.java b/modules/zookeeper/src/test/java/org/apache/ignite/spi/discovery/zk/internal/ZkOperationContextSendAttributesTest.java new file mode 100644 index 0000000000000..c3d33bf581535 --- /dev/null +++ b/modules/zookeeper/src/test/java/org/apache/ignite/spi/discovery/zk/internal/ZkOperationContextSendAttributesTest.java @@ -0,0 +1,32 @@ +/* + * 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.thread.context.OperationContextSendAttributesTest; + +import static org.junit.Assume.assumeTrue; + +/** */ +public class ZkOperationContextSendAttributesTest extends OperationContextSendAttributesTest { + /** {@inheritDoc} */ + @Override protected void doTestOperationContextAttributesPropagation(boolean discovery) throws Exception { + assumeTrue(discovery); + + super.doTestOperationContextAttributesPropagation(true); + } +}