You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by mp...@apache.org on 2026/10/23 05:45:14 UTC
(ignite) branch master updated: IGNITE-29176 Refactored Distributed Operation Context naming and snapshot values iteration (#13711)
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 4f8e676b7b0 IGNITE-29176 Refactored Distributed Operation Context naming and snapshot values iteration (#13711)
4f8e676b7b0 is described below
commit 4f8e676b7b03a9bb1f8b12da0d8ed615c58bd8fa
Author: Mikhail Petrov <32...@users.noreply.github.com>
AuthorDate: Fri Oct 23 08:45:07 2026 +0300
IGNITE-29176 Refactored Distributed Operation Context naming and snapshot values iteration (#13711)
---
.../apache/ignite/internal/GridKernalContext.java | 8 +-
.../ignite/internal/GridKernalContextImpl.java | 6 +-
.../org/apache/ignite/internal/IgniteKernal.java | 16 +--
.../managers/communication/GridIoManager.java | 6 +-
.../wal/reader/StandaloneGridKernalContext.java | 10 +-
.../security/IgniteSecurityProcessor.java | 11 +-
.../security/SecurityContextWrapper.java | 6 +-
.../context/DistributedAttributeIdRegistry.java | 29 ------
.../thread/context/DistributedAttributeKey.java | 71 +++++++++++++
...tcher.java => DistributedOperationContext.java} | 114 +++++++++++----------
.../context/OperationContextSnapshotMessage.java | 73 ++++++-------
.../ignite/spi/discovery/tcp/ClientImpl.java | 4 +-
.../ignite/spi/discovery/tcp/ServerImpl.java | 6 +-
.../ignite/spi/discovery/tcp/TcpDiscoveryImpl.java | 8 +-
...ComputeTaskOperationContextPropagationTest.java | 7 +-
.../OperationContextAttributePropagationTest.java | 18 ++--
.../junits/GridTestKernalContext.java | 4 +-
.../ZkOperationContextAwareCustomMessage.java | 4 +-
.../zk/internal/ZookeeperDiscoveryImpl.java | 10 +-
19 files changed, 229 insertions(+), 182 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
index 0dee5ba7674..ebc321f9606 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
@@ -78,7 +78,7 @@ import org.apache.ignite.internal.processors.subscription.GridInternalSubscripti
import org.apache.ignite.internal.processors.task.GridTaskProcessor;
import org.apache.ignite.internal.processors.timeout.GridTimeoutProcessor;
import org.apache.ignite.internal.suggestions.GridPerformanceSuggestions;
-import org.apache.ignite.internal.thread.context.OperationContextDispatcher;
+import org.apache.ignite.internal.thread.context.DistributedOperationContext;
import org.apache.ignite.internal.util.IgniteExceptionRegistry;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.worker.WorkersRegistry;
@@ -222,11 +222,11 @@ public interface GridKernalContext extends Iterable<GridComponent> {
public MessageFactory messageFactory();
/**
- * Gets the distributed operation context dispatcher.
+ * Gets the distributed operation context.
*
- * @return The distributed operation context dispatcher.
+ * @return The distributed operation context.
*/
- public OperationContextDispatcher operationContextDispatcher();
+ public DistributedOperationContext distributedOperationContext();
/**
* Gets transformation processor.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
index 9892823a091..773efe5a99d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
@@ -100,7 +100,7 @@ import org.apache.ignite.internal.processors.subscription.GridInternalSubscripti
import org.apache.ignite.internal.processors.task.GridTaskProcessor;
import org.apache.ignite.internal.processors.timeout.GridTimeoutProcessor;
import org.apache.ignite.internal.suggestions.GridPerformanceSuggestions;
-import org.apache.ignite.internal.thread.context.OperationContextDispatcher;
+import org.apache.ignite.internal.thread.context.DistributedOperationContext;
import org.apache.ignite.internal.thread.pool.IgniteForkJoinPool;
import org.apache.ignite.internal.util.IgniteExceptionRegistry;
import org.apache.ignite.internal.util.spring.IgniteSpringHelper;
@@ -694,8 +694,8 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
}
/** {@inheritDoc} */
- @Override public OperationContextDispatcher operationContextDispatcher() {
- return grid.operationContextDispatcher();
+ @Override public DistributedOperationContext distributedOperationContext() {
+ return grid.distributedOperationContext();
}
/** {@inheritDoc} */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
index 715e206f923..5d11403186f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
@@ -173,7 +173,7 @@ import org.apache.ignite.internal.suggestions.GridPerformanceSuggestions;
import org.apache.ignite.internal.suggestions.JvmConfigurationSuggestions;
import org.apache.ignite.internal.suggestions.OsConfigurationSuggestions;
import org.apache.ignite.internal.systemview.ConfigurationViewWalker;
-import org.apache.ignite.internal.thread.context.OperationContextDispatcher;
+import org.apache.ignite.internal.thread.context.DistributedOperationContext;
import org.apache.ignite.internal.util.TimeBag;
import org.apache.ignite.internal.util.future.GridCompoundFuture;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
@@ -442,8 +442,8 @@ public class IgniteKernal implements IgniteEx, Externalizable {
/** Core message factory. */
private MessageFactory msgFactory;
- /** Distributed operation context dispatcher. */
- private OperationContextDispatcher operationCtxDispatcher;
+ /** Distributed operation context. */
+ private DistributedOperationContext distrOpCtx;
/**
* No-arg constructor is required by externalization.
@@ -926,7 +926,7 @@ public class IgniteKernal implements IgniteEx, Externalizable {
longJVMPauseDetector
);
- operationCtxDispatcher = new OperationContextDispatcher();
+ distrOpCtx = new DistributedOperationContext();
startProcessor(new DiagnosticProcessor(ctx));
@@ -1144,7 +1144,7 @@ public class IgniteKernal implements IgniteEx, Externalizable {
// All components exept Discovery are started, time to check if maintenance is still needed.
mntcProc.prepareAndExecuteMaintenance();
- operationCtxDispatcher.finishRegistration();
+ distrOpCtx.finishRegistration();
gw.writeLock();
@@ -3016,9 +3016,9 @@ public class IgniteKernal implements IgniteEx, Externalizable {
return msgFactory;
}
- /** @return Distributed operation context dispatcher. */
- OperationContextDispatcher operationContextDispatcher() {
- return operationCtxDispatcher;
+ /** @return Distributed operation context. */
+ DistributedOperationContext distributedOperationContext() {
+ return distrOpCtx;
}
/**
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
index 2f337812d69..90fd123fc33 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
@@ -437,7 +437,7 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Object>>
try {
GridIoMessage msg0 = (GridIoMessage)msg;
- try (Scope ignored = ctx.operationContextDispatcher().restoreSnapshot(msg0.opCtxSnp)) {
+ try (Scope ignored = ctx.distributedOperationContext().restoreSnapshot(msg0.opCtxSnp)) {
onMessage0(nodeId, msg0, msgC);
}
}
@@ -1766,7 +1766,7 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Object>>
long timeout,
boolean skipOnTimeout
) {
- return new GridIoMessage(plc, topic, msg, ordered, timeout, skipOnTimeout, ctx.operationContextDispatcher().createSnapshot());
+ return new GridIoMessage(plc, topic, msg, ordered, timeout, skipOnTimeout, ctx.distributedOperationContext().createSnapshot());
}
/**
@@ -3481,7 +3481,7 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Object>>
assert reserved.get();
for (OrderedMessageContainer mc = msgs.poll(); mc != null; mc = msgs.poll()) {
- try (Scope ignored0 = ctx.operationContextDispatcher().restoreSnapshot(mc.message.opCtxSnp)) {
+ try (Scope ignored0 = ctx.distributedOperationContext().restoreSnapshot(mc.message.opCtxSnp)) {
try {
try {
unmarshalPayload(mc.message);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/reader/StandaloneGridKernalContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/reader/StandaloneGridKernalContext.java
index 835cfb15d14..8f11ca68b96 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/reader/StandaloneGridKernalContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/reader/StandaloneGridKernalContext.java
@@ -102,7 +102,7 @@ import org.apache.ignite.internal.processors.subscription.GridInternalSubscripti
import org.apache.ignite.internal.processors.task.GridTaskProcessor;
import org.apache.ignite.internal.processors.timeout.GridTimeoutProcessor;
import org.apache.ignite.internal.suggestions.GridPerformanceSuggestions;
-import org.apache.ignite.internal.thread.context.OperationContextDispatcher;
+import org.apache.ignite.internal.thread.context.DistributedOperationContext;
import org.apache.ignite.internal.util.IgniteExceptionRegistry;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -177,8 +177,8 @@ public class StandaloneGridKernalContext implements GridKernalContext {
/** Marshaller. */
private final BinaryMarshaller marsh;
- /** Operation context dispacther. */
- private final OperationContextDispatcher opCtxDispatcher = new OperationContextDispatcher();
+ /** Distributed operation context. */
+ private final DistributedOperationContext distrOpCtx = new DistributedOperationContext();
/**
* @param log Logger.
@@ -459,8 +459,8 @@ public class StandaloneGridKernalContext implements GridKernalContext {
}
/** {@inheritDoc} */
- @Override public OperationContextDispatcher operationContextDispatcher() {
- return opCtxDispatcher;
+ @Override public DistributedOperationContext distributedOperationContext() {
+ return distrOpCtx;
}
/** {@inheritDoc} */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/security/IgniteSecurityProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/IgniteSecurityProcessor.java
index b7586b7577a..5e260800132 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/security/IgniteSecurityProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/IgniteSecurityProcessor.java
@@ -32,9 +32,9 @@ import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.processors.security.sandbox.AccessControllerSandbox;
import org.apache.ignite.internal.processors.security.sandbox.IgniteSandbox;
import org.apache.ignite.internal.processors.security.sandbox.NoOpSandbox;
+import org.apache.ignite.internal.thread.context.DistributedAttributeKey;
import org.apache.ignite.internal.thread.context.OperationContext;
import org.apache.ignite.internal.thread.context.OperationContextAttribute;
-import org.apache.ignite.internal.thread.context.OperationContextDispatcher;
import org.apache.ignite.internal.thread.context.Scope;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -56,7 +56,6 @@ import static org.apache.ignite.internal.processors.security.SecurityUtils.IGNIT
import static org.apache.ignite.internal.processors.security.SecurityUtils.MSG_SEC_PROC_CLS_IS_INVALID;
import static org.apache.ignite.internal.processors.security.SecurityUtils.hasSecurityManager;
import static org.apache.ignite.internal.processors.security.SecurityUtils.nodeSecurityContext;
-import static org.apache.ignite.internal.thread.context.DistributedAttributeIdRegistry.SECURITY;
import static org.apache.ignite.plugin.security.SecurityPermission.ADMIN_USER_ACCESS;
import static org.apache.ignite.plugin.security.SecurityPermission.JOIN_AS_SERVER;
@@ -90,11 +89,7 @@ public class IgniteSecurityProcessor extends IgniteSecurityAdapter {
return SANDBOXED_NODES_COUNTER.get() > 0;
}
- /**
- * Attribute that holds local and distributed Security Context.
- *
- * @see OperationContextDispatcher
- */
+ /** */
private static final OperationContextAttribute<SecurityContextWrapper> SEC_CTX_ATTR = OperationContextAttribute.newInstance();
/** Security processor. */
@@ -253,7 +248,7 @@ public class IgniteSecurityProcessor extends IgniteSecurityAdapter {
@Override public void start() throws IgniteCheckedException {
super.start();
- ctx.operationContextDispatcher().registerDistributedAttribute(SECURITY, SEC_CTX_ATTR);
+ ctx.distributedOperationContext().register(DistributedAttributeKey.SECURITY, SEC_CTX_ATTR);
ctx.addNodeAttribute(ATTR_GRID_SEC_PROC_CLASS, secPrc.getClass().getName());
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/security/SecurityContextWrapper.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/SecurityContextWrapper.java
index fbfe58e2f02..77174b02c82 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/security/SecurityContextWrapper.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/security/SecurityContextWrapper.java
@@ -19,16 +19,14 @@ package org.apache.ignite.internal.processors.security;
import java.util.UUID;
import org.apache.ignite.internal.Order;
-import org.apache.ignite.internal.thread.context.DistributedAttributeIdRegistry;
-import org.apache.ignite.internal.thread.context.OperationContextDispatcher;
+import org.apache.ignite.internal.thread.context.DistributedOperationContext;
import org.apache.ignite.plugin.extensions.communication.AbstractMessage;
import org.apache.ignite.plugin.security.SecuritySubject;
/**
* {@link SecurityContext} attribute value holder and message for {@link SecuritySubject}'s id.
*
- * @see OperationContextDispatcher#createSnapshot()
- * @see DistributedAttributeIdRegistry#SECURITY
+ * @see DistributedOperationContext#createSnapshot()
*/
public class SecurityContextWrapper extends AbstractMessage {
/** A value of {@link SecuritySubject#id()} */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/thread/context/DistributedAttributeIdRegistry.java b/modules/core/src/main/java/org/apache/ignite/internal/thread/context/DistributedAttributeIdRegistry.java
deleted file mode 100644
index 7b4d2842272..00000000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/thread/context/DistributedAttributeIdRegistry.java
+++ /dev/null
@@ -1,29 +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.thread.context;
-
-import org.apache.ignite.internal.processors.security.SecurityContext;
-
-/**
- * Declares reserved distributed IDs used to consistently identify {@link OperationContext} attributes across
- * all nodes in the cluster.
- */
-public class DistributedAttributeIdRegistry {
- /** ID Reserved for {@link SecurityContext} propagation. */
- public static final byte SECURITY = 0;
-}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/thread/context/DistributedAttributeKey.java b/modules/core/src/main/java/org/apache/ignite/internal/thread/context/DistributedAttributeKey.java
new file mode 100644
index 00000000000..89e62685c8a
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/thread/context/DistributedAttributeKey.java
@@ -0,0 +1,71 @@
+/*
+ * 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 org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.internal.S;
+
+import static org.apache.ignite.internal.thread.context.OperationContextSnapshotMessage.MAX_ATTRS_CNT;
+
+/**
+ * Represents a key that is used to consistently identify {@link OperationContext} attributes across
+ * all nodes in the cluster.
+ *
+ * @see DistributedOperationContext
+ */
+public final class DistributedAttributeKey {
+ /** Key reserved for the distributed Security Context attribute. */
+ public static final DistributedAttributeKey SECURITY = new DistributedAttributeKey(0);
+
+ /** */
+ @GridToStringInclude
+ private final byte id;
+
+ /** */
+ DistributedAttributeKey(int id) {
+ assert 0 <= id && id < MAX_ATTRS_CNT : "Invalid distributed attribute id [id=" + id + ']';
+
+ this.id = (byte)id;
+ }
+
+ /** @return Cluster-wide id of the attribute. */
+ public byte id() {
+ return id;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (!(o instanceof DistributedAttributeKey))
+ return false;
+
+ return id == ((DistributedAttributeKey)o).id;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return id;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(DistributedAttributeKey.class, this);
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/thread/context/OperationContextDispatcher.java b/modules/core/src/main/java/org/apache/ignite/internal/thread/context/DistributedOperationContext.java
similarity index 54%
rename from modules/core/src/main/java/org/apache/ignite/internal/thread/context/OperationContextDispatcher.java
rename to modules/core/src/main/java/org/apache/ignite/internal/thread/context/DistributedOperationContext.java
index 85b476a7e51..fc22a53c4b1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/thread/context/OperationContextDispatcher.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/thread/context/DistributedOperationContext.java
@@ -18,10 +18,12 @@ package org.apache.ignite.internal.thread.context;
import java.util.Arrays;
import org.apache.ignite.IgniteException;
-import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.jetbrains.annotations.Nullable;
+import static org.apache.ignite.internal.thread.context.OperationContextSnapshotMessage.nextAttributeId;
+import static org.apache.ignite.internal.thread.context.OperationContextSnapshotMessage.set;
+
/**
* Provides the ability to manage {@link OperationContext} attributes in a distributed manner.
*
@@ -32,82 +34,99 @@ import org.jetbrains.annotations.Nullable;
* <p>The implementation relies on a mapping between a distributed identifier and an
* {@link OperationContextAttribute} instance that is consistent across all cluster nodes.</p>
*
- * <p>To enable propagation of an {@link OperationContextAttribute} value across cluster nodes, the
- * attribute must be registered with the {@link #registerDistributedAttribute(int, OperationContextAttribute)} method.
+ * <p>To enable propagation of an {@link OperationContextAttribute} value across cluster nodes, the attribute must be
+ * registered with the {@link #register(DistributedAttributeKey, OperationContextAttribute)}
+ * method.
*
* <p> Note, that the maximum number of distributed attributes to register is currently limited to
- * {@link #MAX_ATTRS_CNT} for implementation reasons.</p>
+ * {@link OperationContextSnapshotMessage#MAX_ATTRS_CNT} for implementation reasons.</p>
*
* @see OperationContext
* @see OperationContextSnapshotMessage
*/
-public class OperationContextDispatcher {
- /** Maximal number of supported distributed attributes. */
- static final byte MAX_ATTRS_CNT = Byte.SIZE;
-
+public class DistributedOperationContext {
/** Registered distributed attributes by their cluster-wide id. */
- private volatile OperationContextAttribute<? extends Message>[] registeredAttrs = new OperationContextAttribute[0];
+ private volatile DistributedAttribute[] distrAttrs = new DistributedAttribute[0];
/** Whether the registration of new distributed attributes is allowed. */
private boolean regFinished;
/**
- * Registers an attribute of {@link OperationContext} with the specified distributed ID.
+ * Registers an attribute of {@link OperationContext} with the specified distributed attribute key.
*
- * <p>The distributed ID is used to consistently identify the attribute across all nodes in the cluster.
- * It must be unique, and its value must be in the range [{@code 0} : {@code Byte.SIZE}).</p>
+ * <p>The key consistently identifies the attribute across all nodes in the cluster and must be unique.</p>
*
* <p>Registered attribute value is automatically captured and propagated between cluster nodes
* during the messages transmission.</p>
*
- * @see DistributedAttributeIdRegistry
+ * @see DistributedAttributeKey
*/
- public synchronized <T extends Message> void registerDistributedAttribute(int id, OperationContextAttribute<T> attr) {
+ public synchronized <T extends Message> void register(DistributedAttributeKey key, OperationContextAttribute<T> attr) {
if (regFinished)
throw new IgniteException("Initialization of distributed operation context attributes has already finished.");
- assert 0 <= id && id < MAX_ATTRS_CNT : "Invalid distributed attributed id [id=" + id + ']';
+ byte id = key.id();
- OperationContextAttribute<? extends Message>[] locRegisteredAttrs = registeredAttrs;
+ if (key.equals(key(id)))
+ throw new IgniteException("Duplicated distributed attribute id [id=" + id + ']');
- OperationContextAttribute<? extends Message>[] copy = Arrays.copyOf(
- locRegisteredAttrs,
- Math.max(locRegisteredAttrs.length, id + 1));
+ DistributedAttribute[] locDistrAttrs = distrAttrs;
- if (copy[id] != null)
- throw new IgniteException("Duplicated distributed attribute id [id=" + id + ']');
+ DistributedAttribute[] copy = Arrays.copyOf(locDistrAttrs, Math.max(locDistrAttrs.length, id + 1));
- copy[id] = attr;
+ copy[id] = new DistributedAttribute(key, attr);
- registeredAttrs = copy;
+ distrAttrs = copy;
+ }
+
+ /** @return Key of the attribute registered under the id, or {@code null} if none is. */
+ public @Nullable DistributedAttributeKey key(int id) {
+ DistributedAttribute[] attrs = distrAttrs;
+
+ if (id < 0 || id >= attrs.length)
+ return null;
+
+ DistributedAttribute attr = attrs[id];
+
+ return attr == null ? null : attrs[id].key;
}
/**
- * Collects the values of all distributed {@link OperationContextAttribute}s registered by this dispatcher.
+ * Collects the values of all registered distributed {@link OperationContextAttribute}s.
*
* @see OperationContext#get(OperationContextAttribute)
*/
public @Nullable OperationContextSnapshotMessage createSnapshot() {
- OperationContextAttribute<? extends Message>[] locRegisteredAttrs = registeredAttrs;
+ DistributedAttribute[] locDistrAttrs = distrAttrs;
- if (locRegisteredAttrs.length == 0)
- return null;
-
- OperationContextSnapshotMessage.Builder snpBuilder = OperationContextSnapshotMessage.Builder.create();
+ Message[] attrs = null;
- for (int id = 0; id < locRegisteredAttrs.length; id++) {
- OperationContextAttribute<? extends Message> attr = locRegisteredAttrs[id];
+ byte idBitmap = 0;
+ int cnt = 0;
- if (attr == null)
+ for (int id = 0; id < locDistrAttrs.length; id++) {
+ if (locDistrAttrs[id] == null)
continue;
+ OperationContextAttribute<? extends Message> attr = locDistrAttrs[id].attr;
+
Message curVal = OperationContext.get(attr);
- if (curVal != attr.initialValue())
- snpBuilder.add(id, curVal);
+ if (curVal == attr.initialValue())
+ continue;
+
+ if (attrs == null)
+ attrs = new Message[locDistrAttrs.length - id];
+
+ attrs[cnt++] = curVal;
+
+ idBitmap = set(idBitmap, id);
}
- return snpBuilder.isEmpty() ? null : snpBuilder.build();
+ if (attrs == null)
+ return null;
+
+ return new OperationContextSnapshotMessage(idBitmap, cnt == attrs.length ? attrs : Arrays.copyOf(attrs, cnt));
}
/** Restores {@link OperationContextAttribute} values received from a remote node. */
@@ -115,27 +134,15 @@ public class OperationContextDispatcher {
if (snp == null)
return OperationContext.Restorer.restoreEmpty();
- OperationContextAttribute<? extends Message>[] locRegisteredAttrs = registeredAttrs;
-
- assert snp.idBitmap != 0;
- assert !F.isEmpty(snp.attrs);
- assert snp.attrs.length <= MAX_ATTRS_CNT;
+ DistributedAttribute[] locDistrAttrs = distrAttrs;
OperationContext.Restorer ctxRestorer = OperationContext.Restorer.create();
- for (byte valIdx = 0, attrId = 0; valIdx < snp.attrs.length; ++valIdx) {
- Message attrVal = snp.attrs[valIdx];
-
- while ((snp.idBitmap & (1 << attrId)) == 0)
- ++attrId;
-
- assert attrId < locRegisteredAttrs.length;
+ for (int attrId = nextAttributeId(snp.idBitmap, -1); attrId >= 0; attrId = nextAttributeId(snp.idBitmap, attrId)) {
+ assert attrId < locDistrAttrs.length;
+ assert locDistrAttrs[attrId] != null;
- OperationContextAttribute<Message> attr = (OperationContextAttribute<Message>)locRegisteredAttrs[attrId++];
-
- assert attr != null;
-
- ctxRestorer.add(attr, attrVal);
+ ctxRestorer.add((OperationContextAttribute<Message>)locDistrAttrs[attrId].attr, snp.getValue(attrId));
}
return ctxRestorer.restore();
@@ -145,4 +152,7 @@ public class OperationContextDispatcher {
public synchronized void finishRegistration() {
regFinished = true;
}
+
+ /** */
+ private record DistributedAttribute(DistributedAttributeKey key, OperationContextAttribute<? extends Message> attr) {}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/thread/context/OperationContextSnapshotMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/thread/context/OperationContextSnapshotMessage.java
index f9a0491f857..0a037492ed8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/thread/context/OperationContextSnapshotMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/thread/context/OperationContextSnapshotMessage.java
@@ -17,25 +17,24 @@
package org.apache.ignite.internal.thread.context;
-import java.util.ArrayList;
-import java.util.List;
import org.apache.ignite.internal.Order;
import org.apache.ignite.plugin.extensions.communication.AbstractMessage;
import org.apache.ignite.plugin.extensions.communication.Message;
-import static org.apache.ignite.internal.thread.context.OperationContextDispatcher.MAX_ATTRS_CNT;
-
/**
* Message for {@link OperationContext} distributed attributes.
*
- * @see OperationContextDispatcher
+ * @see DistributedOperationContext
*/
public class OperationContextSnapshotMessage extends AbstractMessage {
- /** Values of operation context attributes. */
+ /** Maximal number of supported distributed attributes. */
+ static final int MAX_ATTRS_CNT = Byte.SIZE;
+
+ /** Values of operation context attributes, in ascending id order. */
@Order(0)
Message[] attrs;
- /** Bitmap of effective attributes ids. */
+ /** Bitmap of attributes ids. */
@Order(1)
byte idBitmap;
@@ -45,50 +44,44 @@ public class OperationContextSnapshotMessage extends AbstractMessage {
}
/** */
- private OperationContextSnapshotMessage(byte idBitmap, Message[] attrs) {
+ OperationContextSnapshotMessage(byte idBitmap, Message[] attrs) {
+ assert attrs.length == attributesCount(idBitmap);
+
this.attrs = attrs;
this.idBitmap = idBitmap;
}
/** */
- public static class Builder {
- /** */
- private byte bitmap = 0;
-
- /** */
- private List<Message> vals;
+ Message getValue(int attrId) {
+ assert contains(idBitmap, attrId);
- /** */
- private Builder() {
- // No-op.
- }
-
- /** */
- public void add(int attrId, Message attrVal) {
- if (vals == null)
- vals = new ArrayList<>(MAX_ATTRS_CNT / 2);
+ return attrs[valueIndex(attrId)];
+ }
- byte mask = (byte)(1 << attrId);
+ /** */
+ private int valueIndex(int attrId) {
+ return Integer.bitCount(idBitmap & ((1 << attrId) - 1));
+ }
- assert (bitmap & mask) == 0;
+ /** */
+ static int attributesCount(byte idBitmap) {
+ return Integer.bitCount(Byte.toUnsignedInt(idBitmap));
+ }
- vals.add(attrVal);
- bitmap |= mask;
- }
+ /** */
+ static int nextAttributeId(byte idBitmap, int prevId) {
+ int bits = Byte.toUnsignedInt(idBitmap) & (-1 << (prevId + 1));
- /** */
- public boolean isEmpty() {
- return bitmap == 0;
- }
+ return bits == 0 ? -1 : Integer.numberOfTrailingZeros(bits);
+ }
- /** */
- OperationContextSnapshotMessage build() {
- return new OperationContextSnapshotMessage(bitmap, vals.toArray(Message[]::new));
- }
+ /** */
+ static boolean contains(byte idBitmap, int attrId) {
+ return (idBitmap & (1 << attrId)) != 0;
+ }
- /** */
- public static OperationContextSnapshotMessage.Builder create() {
- return new Builder();
- }
+ /** */
+ static byte set(byte idBitmap, int attrId) {
+ return (byte)(idBitmap | (1 << attrId));
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
index 344c97ea74f..ddf9a61102e 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
@@ -1275,7 +1275,7 @@ class ClientImpl extends TcpDiscoveryImpl {
* @param msg Message.
*/
private void sendMessage(TcpDiscoveryAbstractMessage msg) {
- msg.attachOperationContextSnapshot(operationCtxDispatcher.createSnapshot());
+ msg.attachOperationContextSnapshot(distrOpCtx.createSnapshot());
synchronized (mux) {
queue.add(msg);
@@ -2083,7 +2083,7 @@ class ClientImpl extends TcpDiscoveryImpl {
/** */
protected void processDiscoveryMessage(TcpDiscoveryAbstractMessage msg) {
- try (Scope ignored = operationCtxDispatcher.restoreSnapshot(msg.opCtxSnp)) {
+ try (Scope ignored = distrOpCtx.restoreSnapshot(msg.opCtxSnp)) {
processDiscoveryMessage0(msg);
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
index 7ec09237569..9a87615ffbb 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
@@ -2956,7 +2956,7 @@ class ServerImpl extends TcpDiscoveryImpl {
}
if (!fromSocket)
- msg.attachOperationContextSnapshot(operationCtxDispatcher.createSnapshot());
+ msg.attachOperationContextSnapshot(distrOpCtx.createSnapshot());
boolean addFirst = msg.highPriority() && !ignoreHighPriority;
@@ -3183,7 +3183,7 @@ class ServerImpl extends TcpDiscoveryImpl {
if (msg == WAKEUP)
return;
- try (Scope ignored = operationCtxDispatcher.restoreSnapshot(msg.opCtxSnp)) {
+ try (Scope ignored = distrOpCtx.restoreSnapshot(msg.opCtxSnp)) {
processMessage0(msg);
}
}
@@ -6070,7 +6070,7 @@ class ServerImpl extends TcpDiscoveryImpl {
TcpDiscoveryCustomEventMessage msg;
while ((msg = pollPendingCustomMessage()) != null) {
- try (Scope ignored = operationCtxDispatcher.restoreSnapshot(msg.opCtxSnp)) {
+ try (Scope ignored = distrOpCtx.restoreSnapshot(msg.opCtxSnp)) {
processCustomMessage(msg, true);
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryImpl.java
index 0f2324361d5..a3f5a65087e 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryImpl.java
@@ -40,7 +40,7 @@ import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.processors.cache.CacheMetricsSnapshot;
import org.apache.ignite.internal.processors.cluster.CacheMetricsMessage;
import org.apache.ignite.internal.processors.cluster.NodeFullMetricsMessage;
-import org.apache.ignite.internal.thread.context.OperationContextDispatcher;
+import org.apache.ignite.internal.thread.context.DistributedOperationContext;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.LT;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -137,8 +137,8 @@ abstract class TcpDiscoveryImpl {
}
};
- /** Distributed operation context dispatcher. */
- protected final OperationContextDispatcher operationCtxDispatcher;
+ /** Distributed operation context. */
+ protected final DistributedOperationContext distrOpCtx;
/**
* @param spi Adapter.
@@ -150,7 +150,7 @@ abstract class TcpDiscoveryImpl {
ctx = ((IgniteEx)spi.ignite()).context();
- operationCtxDispatcher = ctx.operationContextDispatcher();
+ distrOpCtx = ctx.distributedOperationContext();
}
/**
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/thread/context/ComputeTaskOperationContextPropagationTest.java b/modules/core/src/test/java/org/apache/ignite/internal/thread/context/ComputeTaskOperationContextPropagationTest.java
index dd6bff0c80f..4a36894f8dd 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/thread/context/ComputeTaskOperationContextPropagationTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/thread/context/ComputeTaskOperationContextPropagationTest.java
@@ -55,7 +55,7 @@ import org.junit.Test;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.ignite.events.EventType.EVT_JOB_REJECTED;
import static org.apache.ignite.internal.TestRecordingCommunicationSpi.spi;
-import static org.apache.ignite.internal.thread.context.OperationContextDispatcher.MAX_ATTRS_CNT;
+import static org.apache.ignite.internal.thread.context.OperationContextSnapshotMessage.MAX_ATTRS_CNT;
/** */
public class ComputeTaskOperationContextPropagationTest extends GridCommonAbstractTest {
@@ -421,6 +421,9 @@ public class ComputeTaskOperationContextPropagationTest extends GridCommonAbstra
/** */
private static class TestIgniteComponent extends AbstractTestPluginProvider {
+ /** */
+ private static final DistributedAttributeKey USR_KEY = new DistributedAttributeKey(MAX_ATTRS_CNT - 1);
+
/** {@inheritDoc} */
@Override public String name() {
return "TestComputeTaskOperationContextAttributeRegistrator";
@@ -428,7 +431,7 @@ public class ComputeTaskOperationContextPropagationTest extends GridCommonAbstra
/** {@inheritDoc} */
@Override public void start(PluginContext ctx) {
- ((IgniteEx)ctx.grid()).context().operationContextDispatcher().registerDistributedAttribute(MAX_ATTRS_CNT - 1, USR_ATTR);
+ ((IgniteEx)ctx.grid()).context().distributedOperationContext().register(USR_KEY, USR_ATTR);
}
}
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/thread/context/OperationContextAttributePropagationTest.java b/modules/core/src/test/java/org/apache/ignite/internal/thread/context/OperationContextAttributePropagationTest.java
index ea4978a711e..e39cbbb5dea 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/thread/context/OperationContextAttributePropagationTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/thread/context/OperationContextAttributePropagationTest.java
@@ -59,7 +59,7 @@ import static org.apache.ignite.internal.thread.context.OperationContextAttribut
import static org.apache.ignite.internal.thread.context.OperationContextAttributePropagationTest.TestIgniteComponent.USR_ATTR;
import static org.apache.ignite.internal.thread.context.OperationContextAttributePropagationTest.TestIgniteComponent.discoveryDataExchangeStartedLatch;
import static org.apache.ignite.internal.thread.context.OperationContextAttributePropagationTest.TestIgniteComponent.discoveryDataExchangeUnblockedLatch;
-import static org.apache.ignite.internal.thread.context.OperationContextDispatcher.MAX_ATTRS_CNT;
+import static org.apache.ignite.internal.thread.context.OperationContextSnapshotMessage.MAX_ATTRS_CNT;
import static org.apache.ignite.testframework.GridTestUtils.assertThrows;
import static org.apache.ignite.testframework.GridTestUtils.assertThrowsAnyCause;
import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
@@ -73,6 +73,12 @@ public class OperationContextAttributePropagationTest extends GridCommonAbstract
"Delay custom message processing, there are joining nodes"
).build();
+ /** */
+ public static final DistributedAttributeKey PTR_KEY = new DistributedAttributeKey(0);
+
+ /** */
+ public static final DistributedAttributeKey USR_KEY = new DistributedAttributeKey(MAX_ATTRS_CNT - 1);
+
/** */
private volatile Consumer<Integer> discoMsgLsnr;
@@ -223,7 +229,7 @@ public class OperationContextAttributePropagationTest extends GridCommonAbstract
assertThrows(
null,
- () -> grid(0).context().operationContextDispatcher().registerDistributedAttribute(1, null),
+ () -> grid(0).context().distributedOperationContext().register(PTR_KEY, null),
IgniteException.class,
"Initialization of distributed operation context attributes has already finished"
);
@@ -348,7 +354,7 @@ public class OperationContextAttributePropagationTest extends GridCommonAbstract
}
}
- /** Prevents {@link ClusterNode#isLocal()} to be negative. */
+ /** */
private ClusterNode node(Ignite from, Ignite to) {
return from.cluster().node(((IgniteEx)to).localNode().id());
}
@@ -404,13 +410,13 @@ public class OperationContextAttributePropagationTest extends GridCommonAbstract
@Override public void start(PluginContext ctx) {
kctx = ((IgniteEx)ctx.grid()).context();
- kctx.operationContextDispatcher().registerDistributedAttribute(MAX_ATTRS_CNT - 1, USR_ATTR);
- kctx.operationContextDispatcher().registerDistributedAttribute(0, PTR_ATTR);
+ kctx.distributedOperationContext().register(USR_KEY, USR_ATTR);
+ kctx.distributedOperationContext().register(PTR_KEY, PTR_ATTR);
assertThrowsAnyCause(
log,
() -> {
- kctx.operationContextDispatcher().registerDistributedAttribute(MAX_ATTRS_CNT - 1, PTR_ATTR);
+ kctx.distributedOperationContext().register(USR_KEY, PTR_ATTR);
return null;
}, IgniteException.class,
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridTestKernalContext.java b/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridTestKernalContext.java
index a2528c205b5..c9064f1c1cc 100644
--- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridTestKernalContext.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridTestKernalContext.java
@@ -34,7 +34,7 @@ import org.apache.ignite.internal.processors.metric.GridMetricManager;
import org.apache.ignite.internal.processors.plugin.IgnitePluginProcessor;
import org.apache.ignite.internal.processors.resource.GridResourceProcessor;
import org.apache.ignite.internal.processors.subscription.GridInternalSubscriptionProcessor;
-import org.apache.ignite.internal.thread.context.OperationContextDispatcher;
+import org.apache.ignite.internal.thread.context.DistributedOperationContext;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.plugin.PluginProvider;
@@ -79,7 +79,7 @@ public class GridTestKernalContext extends GridKernalContextImpl {
GridTestUtils.setFieldValue(grid(), "cfg", config());
GridTestUtils.setFieldValue(grid(), "ctx", this);
- GridTestUtils.setFieldValue(grid(), "operationCtxDispatcher", new OperationContextDispatcher());
+ GridTestUtils.setFieldValue(grid(), "distrOpCtx", new DistributedOperationContext());
config().setGridLogger(log);
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
index 75252caf7e6..f9e592a0d2c 100644
--- 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
@@ -18,8 +18,8 @@
package org.apache.ignite.spi.discovery.zk.internal;
import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.thread.context.DistributedOperationContext;
import org.apache.ignite.internal.thread.context.OperationContext;
-import org.apache.ignite.internal.thread.context.OperationContextDispatcher;
import org.apache.ignite.internal.thread.context.OperationContextSnapshotMessage;
import org.apache.ignite.plugin.extensions.communication.AbstractMessage;
import org.apache.ignite.plugin.extensions.communication.MessageFactory;
@@ -37,7 +37,7 @@ import org.jetbrains.annotations.Nullable;
* {@link OperationContext} with all the messages. In {@link ZookeeperDiscoverySpi} with from-Ignite
* {@link DiscoverySpiCustomMessage} only.</p>
*
- * @see OperationContextDispatcher
+ * @see DistributedOperationContext
* @see ZookeeperDiscoverySpi#sendCustomEvent(DiscoverySpiCustomMessage)
*/
public class ZkOperationContextAwareCustomMessage extends AbstractMessage implements DiscoverySpiCustomMessage {
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 e8997b68759..faadb17cdd9 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
@@ -66,7 +66,7 @@ import org.apache.ignite.internal.IgnitionEx;
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.DistributedOperationContext;
import org.apache.ignite.internal.thread.context.OperationContextSnapshotMessage;
import org.apache.ignite.internal.thread.context.Scope;
import org.apache.ignite.internal.thread.pool.IgniteThreadPoolExecutor;
@@ -229,7 +229,7 @@ public class ZookeeperDiscoveryImpl {
private final DiscoveryMessageParser msgParser;
/** */
- private final OperationContextDispatcher opCtxDispatcher;
+ private final DistributedOperationContext distrOpCtx;
/**
* @param spi Discovery SPI.
@@ -282,7 +282,7 @@ public class ZookeeperDiscoveryImpl {
msgParser = new DiscoveryMessageParser(msgFactory, ((IgniteEx)spi.ignite()).context());
- opCtxDispatcher = ((IgniteEx)spi.ignite()).context().operationContextDispatcher();
+ distrOpCtx = ((IgniteEx)spi.ignite()).context().distributedOperationContext();
}
/**
@@ -667,7 +667,7 @@ public class ZookeeperDiscoveryImpl {
/** */
public void sendCustomEvent(DiscoverySpiCustomMessage msg) {
- OperationContextSnapshotMessage opCtxSnp = opCtxDispatcher.createSnapshot();
+ OperationContextSnapshotMessage opCtxSnp = distrOpCtx.createSnapshot();
if (opCtxSnp != null)
sendCustomMessage(new ZkOperationContextAwareCustomMessage(msg, opCtxSnp));
@@ -3543,7 +3543,7 @@ public class ZookeeperDiscoveryImpl {
IgniteFuture<?> fut;
- try (Scope ignored = opCtxDispatcher.restoreSnapshot(opCtxSnp)) {
+ try (Scope ignored = distrOpCtx.restoreSnapshot(opCtxSnp)) {
fut = lsnr.onDiscovery(
new DiscoveryNotification(
DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT,