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 e8b7486f2ef IGNITE-28915 Fixed Operation Context propagation for
postponed messages (#13396)
e8b7486f2ef is described below
commit e8b7486f2efc510213bf7a9cf3e82a06c46bbfab
Author: Mikhail Petrov <[email protected]>
AuthorDate: Mon Jul 27 19:04:57 2026 +0300
IGNITE-28915 Fixed Operation Context propagation for postponed messages
(#13396)
---
.../internal/thread/context/OperationContext.java | 97 ++++--
.../ignite/internal/CoreMessagesProvider.java | 4 +-
.../managers/communication/GridIoManager.java | 24 +-
.../managers/communication/GridIoMessage.java | 9 +-
.../internal/processors/pool/PoolProcessor.java | 2 +-
.../security/IgniteSecurityProcessor.java | 2 +-
.../security/SecurityContextWrapper.java | 6 +-
...ry.java => DistributedAttributeIdRegistry.java} | 4 +-
.../thread/context/OperationContextDispatcher.java | 53 ++-
...e.java => OperationContextSnapshotMessage.java} | 52 ++-
.../ignite/spi/discovery/tcp/ClientImpl.java | 363 +++++++++++----------
.../ignite/spi/discovery/tcp/ServerImpl.java | 21 +-
.../tcp/messages/TcpDiscoveryAbstractMessage.java | 20 +-
.../datastreamer/DataStreamerImplSelfTest.java | 3 +-
.../OperationContextAttributePropagationTest.java | 165 +++++++++-
.../ZkOperationContextAwareCustomMessage.java | 16 +-
.../zk/internal/ZookeeperDiscoveryImpl.java | 14 +-
17 files changed, 553 insertions(+), 302 deletions(-)
diff --git
a/modules/commons/src/main/java/org/apache/ignite/internal/thread/context/OperationContext.java
b/modules/commons/src/main/java/org/apache/ignite/internal/thread/context/OperationContext.java
index 4a8f556781c..2b98393d25d 100644
---
a/modules/commons/src/main/java/org/apache/ignite/internal/thread/context/OperationContext.java
+++
b/modules/commons/src/main/java/org/apache/ignite/internal/thread/context/OperationContext.java
@@ -113,7 +113,7 @@ public class OperationContext {
OperationContextAttribute<T1> attr1, T1 val1,
OperationContextAttribute<T2> attr2, T2 val2
) {
- return ContextUpdater.create().set(attr1, val1).set(attr2,
val2).apply();
+ return Updater.create().set(attr1, val1).set(attr2, val2).apply();
}
/**
@@ -136,7 +136,7 @@ public class OperationContext {
OperationContextAttribute<T2> attr2, T2 val2,
OperationContextAttribute<T3> attr3, T3 val3
) {
- return ContextUpdater.create().set(attr1, val1).set(attr2,
val2).set(attr3, val3).apply();
+ return Updater.create().set(attr1, val1).set(attr2, val2).set(attr3,
val3).apply();
}
/**
@@ -322,49 +322,96 @@ public class OperationContext {
}
/** Allows to change multiple attribute values in a single update
operation and skip updates that changes nothing. */
- static class ContextUpdater {
+ static class Updater extends AttributeCollector {
/** */
- private static final int INIT_UPDATES_CAPACITY = 3;
+ private Updater(OperationContext ctx) {
+ super(ctx);
+ }
- /** */
- private final OperationContext ctx;
+ /** */
+ <T> Updater set(OperationContextAttribute<T> attr, T val) {
+ if (ctx.getInternal(attr) == val)
+ return this;
+
+ add(attr, val);
+
+ return this;
+ }
/** */
- private List<AttributeValueHolder<?>> updates;
+ Scope apply() {
+ return isEmpty() ? NOOP_SCOPE :
ctx.applyAttributeUpdates(toArray());
+ }
/** */
- private ContextUpdater(OperationContext ctx) {
- this.ctx = ctx;
+ static Updater create() {
+ return new Updater(INSTANCE.get());
}
+ }
- /** */
- <T> ContextUpdater set(OperationContextAttribute<T> attr, T val) {
- if (ctx.getInternal(attr) == val)
- return this;
+ /**
+ * Provides the ability to restore the {@link OperationContext} state from
the collected attribute values.
+ * Attributes that are not specified are treated as unset.
+ *
+ * <p>Internally, this class constructs compressed {@link
OperationContextSnapshot} and restores it for the calling
+ * thread when {@link #restore()} or {@link #restoreEmpty()} is
invoked.</p>
+ *
+ * @see OperationContext#restoreSnapshot(OperationContextSnapshot)
+ */
+ static class Restorer extends AttributeCollector {
+ /** */
+ private Restorer(OperationContext ctx) {
+ super(ctx);
+ }
- if (updates == null)
- updates = new ArrayList<>(INIT_UPDATES_CAPACITY);
+ /** */
+ Scope restore() {
+ return ctx.restoreSnapshotInternal(isEmpty() ? null : ctx.new
Update(toArray(), null));
+ }
- updates.add(new AttributeValueHolder<>(attr, val));
+ /** */
+ static Restorer create() {
+ return new Restorer(INSTANCE.get());
+ }
- return this;
+ /** */
+ static Scope restoreEmpty() {
+ return INSTANCE.get().restoreSnapshotInternal(null);
}
+ }
+ /** */
+ private abstract static class AttributeCollector {
/** */
- Scope apply() {
- if (F.isEmpty(updates))
- return NOOP_SCOPE;
+ private static final int INIT_CAPACITY = 3;
- AttributeValueHolder<?>[] sealedUpdates = new
AttributeValueHolder[updates.size()];
+ /** */
+ protected final OperationContext ctx;
- updates.toArray(sealedUpdates);
+ /** */
+ private List<AttributeValueHolder<?>> attrs;
+
+ /** */
+ protected AttributeCollector(OperationContext ctx) {
+ this.ctx = ctx;
+ }
- return ctx.applyAttributeUpdates(sealedUpdates);
+ /** */
+ <T> void add(OperationContextAttribute<T> attr, T val) {
+ if (attrs == null)
+ attrs = new ArrayList<>(INIT_CAPACITY);
+
+ attrs.add(new AttributeValueHolder<>(attr, val));
+ }
+
+ /** */
+ protected boolean isEmpty() {
+ return F.isEmpty(attrs);
}
/** */
- static ContextUpdater create() {
- return new ContextUpdater(INSTANCE.get());
+ protected AttributeValueHolder<?>[] toArray() {
+ return attrs.toArray(AttributeValueHolder<?>[]::new);
}
}
}
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 69e92e29ce0..bba4c1a6ca6 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
@@ -259,7 +259,7 @@ import
org.apache.ignite.internal.processors.service.ServiceSingleNodeDeployment
import
org.apache.ignite.internal.processors.service.ServiceSingleNodeDeploymentResultBatch;
import org.apache.ignite.internal.processors.service.ServiceTopology;
import
org.apache.ignite.internal.processors.service.ServiceUndeploymentRequest;
-import org.apache.ignite.internal.thread.context.OperationContextMessage;
+import
org.apache.ignite.internal.thread.context.OperationContextSnapshotMessage;
import org.apache.ignite.internal.util.GridByteArrayList;
import org.apache.ignite.internal.util.GridIntList;
import org.apache.ignite.internal.util.GridPartitionStateMap;
@@ -712,7 +712,7 @@ public class CoreMessagesProvider extends
AbstractMarshallableMessageFactoryProv
// [13400 - 13500]: Operation context messages.
msgIdx = 13400;
- withNoSchema(OperationContextMessage.class);
+ withNoSchema(OperationContextSnapshotMessage.class);
withNoSchema(SecurityContextWrapper.class);
// [13600 - 13700]: Rolling Upgrade messages.
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 ec010801457..40ecd2c6d26 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
@@ -451,7 +451,7 @@ public class GridIoManager extends
GridManagerAdapter<CommunicationSpi<Object>>
try {
GridIoMessage msg0 = (GridIoMessage)msg;
- try (Scope ignored =
ctx.operationContextDispatcher().restoreRemoteAttributeValues(msg0.opCtxMsg)) {
+ try (Scope ignored =
ctx.operationContextDispatcher().restoreSnapshot(msg0.opCtxSnp)) {
onMessage0(nodeId, msg0, msgC);
}
}
@@ -2027,13 +2027,7 @@ public class GridIoManager extends
GridManagerAdapter<CommunicationSpi<Object>>
long timeout,
boolean skipOnTimeout
) {
- GridIoMessage res;
-
- res = new GridIoMessage(plc, topic, msg, ordered, timeout,
skipOnTimeout);
-
- res.opCtxMsg =
ctx.operationContextDispatcher().collectDistributedAttributeValues();
-
- return res;
+ return new GridIoMessage(plc, topic, msg, ordered, timeout,
skipOnTimeout, ctx.operationContextDispatcher().createSnapshot());
}
/**
@@ -3782,12 +3776,14 @@ public class GridIoManager extends
GridManagerAdapter<CommunicationSpi<Object>>
assert reserved.get();
for (OrderedMessageContainer mc = msgs.poll(); mc != null; mc =
msgs.poll()) {
- try {
- invokeListener(plc, lsnr, nodeId, mc.message.message());
- }
- finally {
- if (mc.closure != null)
- mc.closure.run();
+ try (Scope ignored0 =
ctx.operationContextDispatcher().restoreSnapshot(mc.message.opCtxSnp)) {
+ try {
+ invokeListener(plc, lsnr, nodeId,
mc.message.message());
+ }
+ finally {
+ if (mc.closure != null)
+ mc.closure.run();
+ }
}
}
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessage.java
b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessage.java
index 6493e975bba..395540c89b2 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessage.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessage.java
@@ -22,7 +22,7 @@ import org.apache.ignite.internal.GridTopicMessage;
import org.apache.ignite.internal.Order;
import org.apache.ignite.internal.processors.cache.GridCacheMessage;
import org.apache.ignite.internal.processors.datastreamer.DataStreamerRequest;
-import org.apache.ignite.internal.thread.context.OperationContextMessage;
+import
org.apache.ignite.internal.thread.context.OperationContextSnapshotMessage;
import org.apache.ignite.internal.util.nio.GridNioServer.MessageWrapper;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -64,7 +64,7 @@ public class GridIoMessage implements Message, MessageWrapper
{
/** Effective operation context attributes to propagate. */
@Order(6)
@GridToStringInclude
- public @Nullable OperationContextMessage opCtxMsg;
+ @Nullable OperationContextSnapshotMessage opCtxSnp;
/**
* Default constructor.
@@ -80,6 +80,7 @@ public class GridIoMessage implements Message, MessageWrapper
{
* @param ordered Message ordered flag.
* @param timeout Timeout.
* @param skipOnTimeout Whether message can be skipped on timeout.
+ * @param opCtxSnp Operation Context snapshot.
*/
public GridIoMessage(
byte plc,
@@ -87,7 +88,8 @@ public class GridIoMessage implements Message, MessageWrapper
{
Message msg,
boolean ordered,
long timeout,
- boolean skipOnTimeout
+ boolean skipOnTimeout,
+ @Nullable OperationContextSnapshotMessage opCtxSnp
) {
assert topic != null;
assert msg != null;
@@ -98,6 +100,7 @@ public class GridIoMessage implements Message,
MessageWrapper {
this.ordered = ordered;
this.timeout = timeout;
this.skipOnTimeout = skipOnTimeout;
+ this.opCtxSnp = opCtxSnp;
}
/**
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/pool/PoolProcessor.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/pool/PoolProcessor.java
index 1cc0d4ff023..27d9642a96d 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/pool/PoolProcessor.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/pool/PoolProcessor.java
@@ -273,7 +273,7 @@ public class PoolProcessor extends GridProcessorAdapter {
throw new IgniteException("Failed to register IO
executor pool because its ID as " +
"already used: " + id);
- extPools[id] = ctx.security().enabled() ?
OperationContextAwareIoPool.wrap(ex) : ex;
+ extPools[id] = OperationContextAwareIoPool.wrap(ex);
}
}
}
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 a997e9e6a04..b7586b7577a 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
@@ -56,7 +56,7 @@ 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.DistributedAttributeRegistry.SECURITY;
+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;
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 79e869756c4..a1feb8488a3 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,7 +19,7 @@ package org.apache.ignite.internal.processors.security;
import java.util.UUID;
import org.apache.ignite.internal.Order;
-import org.apache.ignite.internal.thread.context.DistributedAttributeRegistry;
+import
org.apache.ignite.internal.thread.context.DistributedAttributeIdRegistry;
import org.apache.ignite.internal.thread.context.OperationContextDispatcher;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.security.SecuritySubject;
@@ -27,8 +27,8 @@ import org.apache.ignite.plugin.security.SecuritySubject;
/**
* {@link SecurityContext} attribute value holder and message for {@link
SecuritySubject}'s id.
*
- * @see OperationContextDispatcher#collectDistributedAttributeValues()
- * @see DistributedAttributeRegistry#SECURITY
+ * @see OperationContextDispatcher#createSnapshot()
+ * @see DistributedAttributeIdRegistry#SECURITY
*/
public class SecurityContextWrapper implements Message {
/** A value of {@link SecuritySubject#id()} */
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/thread/context/DistributedAttributeRegistry.java
b/modules/core/src/main/java/org/apache/ignite/internal/thread/context/DistributedAttributeIdRegistry.java
similarity index 91%
rename from
modules/core/src/main/java/org/apache/ignite/internal/thread/context/DistributedAttributeRegistry.java
rename to
modules/core/src/main/java/org/apache/ignite/internal/thread/context/DistributedAttributeIdRegistry.java
index 870f8e1a566..7b4d2842272 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/thread/context/DistributedAttributeRegistry.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/thread/context/DistributedAttributeIdRegistry.java
@@ -23,7 +23,7 @@ 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 DistributedAttributeRegistry {
- /** Reserved for {@link SecurityContext} propagation. */
+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/OperationContextDispatcher.java
b/modules/core/src/main/java/org/apache/ignite/internal/thread/context/OperationContextDispatcher.java
index e38b9fd18dd..85b476a7e51 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/OperationContextDispatcher.java
@@ -16,9 +16,7 @@
*/
package org.apache.ignite.internal.thread.context;
-import java.util.ArrayList;
import java.util.Arrays;
-import java.util.List;
import org.apache.ignite.IgniteException;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.plugin.extensions.communication.Message;
@@ -41,7 +39,7 @@ import org.jetbrains.annotations.Nullable;
* {@link #MAX_ATTRS_CNT} for implementation reasons.</p>
*
* @see OperationContext
- * @see OperationContextMessage
+ * @see OperationContextSnapshotMessage
*/
public class OperationContextDispatcher {
/** Maximal number of supported distributed attributes. */
@@ -62,7 +60,7 @@ public class OperationContextDispatcher {
* <p>Registered attribute value is automatically captured and propagated
between cluster nodes
* during the messages transmission.</p>
*
- * @see DistributedAttributeRegistry
+ * @see DistributedAttributeIdRegistry
*/
public synchronized <T extends Message> void
registerDistributedAttribute(int id, OperationContextAttribute<T> attr) {
if (regFinished)
@@ -89,14 +87,13 @@ public class OperationContextDispatcher {
*
* @see OperationContext#get(OperationContextAttribute)
*/
- public @Nullable OperationContextMessage
collectDistributedAttributeValues() {
+ public @Nullable OperationContextSnapshotMessage createSnapshot() {
OperationContextAttribute<? extends Message>[] locRegisteredAttrs =
registeredAttrs;
if (locRegisteredAttrs.length == 0)
return null;
- byte bitmap = 0;
- List<Message> vals = null;
+ OperationContextSnapshotMessage.Builder snpBuilder =
OperationContextSnapshotMessage.Builder.create();
for (int id = 0; id < locRegisteredAttrs.length; id++) {
OperationContextAttribute<? extends Message> attr =
locRegisteredAttrs[id];
@@ -106,40 +103,30 @@ public class OperationContextDispatcher {
Message curVal = OperationContext.get(attr);
- if (curVal == attr.initialValue())
- continue;
-
- if (vals == null)
- vals = new ArrayList<>(MAX_ATTRS_CNT / 2);
-
- byte mask = (byte)(1 << id);
-
- assert (bitmap & mask) == 0;
-
- vals.add(curVal);
- bitmap |= mask;
+ if (curVal != attr.initialValue())
+ snpBuilder.add(id, curVal);
}
- return bitmap == 0 ? null : new OperationContextMessage(bitmap,
vals.toArray(Message[]::new));
+ return snpBuilder.isEmpty() ? null : snpBuilder.build();
}
- /** Restores distributed {@link OperationContextAttribute} values received
from a remote node. */
- public Scope restoreRemoteAttributeValues(@Nullable
OperationContextMessage msg) {
- if (msg == null)
- return Scope.NOOP_SCOPE;
+ /** Restores {@link OperationContextAttribute} values received from a
remote node. */
+ public Scope restoreSnapshot(@Nullable OperationContextSnapshotMessage
snp) {
+ if (snp == null)
+ return OperationContext.Restorer.restoreEmpty();
OperationContextAttribute<? extends Message>[] locRegisteredAttrs =
registeredAttrs;
- assert msg.idBitmap != 0;
- assert !F.isEmpty(msg.attrs);
- assert msg.attrs.length <= MAX_ATTRS_CNT;
+ assert snp.idBitmap != 0;
+ assert !F.isEmpty(snp.attrs);
+ assert snp.attrs.length <= MAX_ATTRS_CNT;
- OperationContext.ContextUpdater updater =
OperationContext.ContextUpdater.create();
+ OperationContext.Restorer ctxRestorer =
OperationContext.Restorer.create();
- for (byte valIdx = 0, attrId = 0; valIdx < msg.attrs.length; ++valIdx)
{
- Message curVal = msg.attrs[valIdx];
+ for (byte valIdx = 0, attrId = 0; valIdx < snp.attrs.length; ++valIdx)
{
+ Message attrVal = snp.attrs[valIdx];
- while ((msg.idBitmap & (1 << attrId)) == 0)
+ while ((snp.idBitmap & (1 << attrId)) == 0)
++attrId;
assert attrId < locRegisteredAttrs.length;
@@ -148,10 +135,10 @@ public class OperationContextDispatcher {
assert attr != null;
- updater.set(attr, curVal);
+ ctxRestorer.add(attr, attrVal);
}
- return updater.apply();
+ return ctxRestorer.restore();
}
/** Restricts further registration of distributed attributes. */
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/thread/context/OperationContextMessage.java
b/modules/core/src/main/java/org/apache/ignite/internal/thread/context/OperationContextSnapshotMessage.java
similarity index 52%
rename from
modules/core/src/main/java/org/apache/ignite/internal/thread/context/OperationContextMessage.java
rename to
modules/core/src/main/java/org/apache/ignite/internal/thread/context/OperationContextSnapshotMessage.java
index 01dea73efd5..08fa263efa6 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/thread/context/OperationContextMessage.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/thread/context/OperationContextSnapshotMessage.java
@@ -17,15 +17,19 @@
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.Message;
+import static
org.apache.ignite.internal.thread.context.OperationContextDispatcher.MAX_ATTRS_CNT;
+
/**
* Message for {@link OperationContext} distributed attributes.
*
* @see OperationContextDispatcher
*/
-public class OperationContextMessage implements Message {
+public class OperationContextSnapshotMessage implements Message {
/** Values of operation context attributes. */
@Order(0)
Message[] attrs;
@@ -35,13 +39,55 @@ public class OperationContextMessage implements Message {
byte idBitmap;
/** Empty constructor for serialization purposes. */
- public OperationContextMessage() {
+ public OperationContextSnapshotMessage() {
// No-op.
}
/** */
- public OperationContextMessage(byte idBitmap, Message[] attrs) {
+ private OperationContextSnapshotMessage(byte idBitmap, Message[] attrs) {
this.attrs = attrs;
this.idBitmap = idBitmap;
}
+
+ /** */
+ public static class Builder {
+ /** */
+ private byte bitmap = 0;
+
+ /** */
+ private List<Message> vals;
+
+ /** */
+ private Builder() {
+ // No-op.
+ }
+
+ /** */
+ public void add(int attrId, Message attrVal) {
+ if (vals == null)
+ vals = new ArrayList<>(MAX_ATTRS_CNT / 2);
+
+ byte mask = (byte)(1 << attrId);
+
+ assert (bitmap & mask) == 0;
+
+ vals.add(attrVal);
+ bitmap |= mask;
+ }
+
+ /** */
+ public boolean isEmpty() {
+ return bitmap == 0;
+ }
+
+ /** */
+ OperationContextSnapshotMessage build() {
+ return new OperationContextSnapshotMessage(bitmap,
vals.toArray(Message[]::new));
+ }
+
+ /** */
+ public static OperationContextSnapshotMessage.Builder create() {
+ return new Builder();
+ }
+ }
}
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 8cd65e7d356..918a29bab42 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
@@ -1278,7 +1278,7 @@ class ClientImpl extends TcpDiscoveryImpl {
* @param msg Message.
*/
private void sendMessage(TcpDiscoveryAbstractMessage msg) {
- msg.opCtxMsg =
operationCtxDispatcher.collectDistributedAttributeValues();
+
msg.attachOperationContextSnapshot(operationCtxDispatcher.createSnapshot());
synchronized (mux) {
queue.add(msg);
@@ -1727,245 +1727,241 @@ class ClientImpl extends TcpDiscoveryImpl {
blockingSectionEnd();
}
- TcpDiscoveryAbstractMessage dm = msg instanceof
TcpDiscoveryAbstractMessage
- ? (TcpDiscoveryAbstractMessage)msg
- : null;
+ if (msg instanceof JoinTimeout) {
+ int joinCnt0 = ((JoinTimeout)msg).joinCnt;
- try (Scope ignored =
operationCtxDispatcher.restoreRemoteAttributeValues(dm == null ? null :
dm.opCtxMsg)) {
- if (msg instanceof JoinTimeout) {
- int joinCnt0 = ((JoinTimeout)msg).joinCnt;
+ if (joinCnt == joinCnt0) {
+ if (state == STARTING) {
+ joinError(new IgniteSpiException("Join process
timed out, did not receive response for " +
+ "join request (consider increasing
'joinTimeout' configuration property) " +
+ "[joinTimeout=" + spi.joinTimeout + ",
sock=" + currSock + ']'));
- if (joinCnt == joinCnt0) {
- if (state == STARTING) {
- joinError(new IgniteSpiException("Join
process timed out, did not receive response for " +
- "join request (consider increasing
'joinTimeout' configuration property) " +
- "[joinTimeout=" + spi.joinTimeout + ",
sock=" + currSock + ']'));
-
- break;
- }
- else if (state == DISCONNECTED) {
- if (log.isDebugEnabled())
- log.debug("Failed to reconnect, local
node segmented " +
- "[joinTimeout=" + spi.joinTimeout
+ ']');
+ break;
+ }
+ else if (state == DISCONNECTED) {
+ if (log.isDebugEnabled())
+ log.debug("Failed to reconnect, local node
segmented " +
+ "[joinTimeout=" + spi.joinTimeout +
']');
- state = SEGMENTED;
+ state = SEGMENTED;
- notifyDiscovery(
- EVT_NODE_SEGMENTED, topVer, locNode,
allVisibleNodes());
- }
+ notifyDiscovery(
+ EVT_NODE_SEGMENTED, topVer, locNode,
allVisibleNodes(), null);
}
}
- else if (msg == SPI_STOP) {
- boolean connected = state == CONNECTED;
+ }
+ else if (msg == SPI_STOP) {
+ boolean connected = state == CONNECTED;
- state = STOPPED;
+ state = STOPPED;
- assert spi.getSpiContext().isStopping();
+ assert spi.getSpiContext().isStopping();
- if (connected && currSock != null) {
- TcpDiscoveryNodeLeftMessage leftMsg = new
TcpDiscoveryNodeLeftMessage(getLocalNodeId());
+ if (connected && currSock != null) {
+ TcpDiscoveryNodeLeftMessage leftMsg = new
TcpDiscoveryNodeLeftMessage(getLocalNodeId());
- leftMsg.client(true);
+ leftMsg.client(true);
- sockWriter.sendMessage(leftMsg);
- }
- else
- leaveLatch.countDown();
+ sockWriter.sendMessage(leftMsg);
}
- else if (msg == SPI_RECONNECT) {
- if (state == CONNECTED) {
- if (reconnector != null) {
- reconnector.cancel();
- reconnector.join();
+ else
+ leaveLatch.countDown();
+ }
+ else if (msg == SPI_RECONNECT) {
+ if (state == CONNECTED) {
+ if (reconnector != null) {
+ reconnector.cancel();
+ reconnector.join();
- reconnector = null;
- }
+ reconnector = null;
+ }
- sockWriter.forceLeave();
- sockReader.forceStopRead();
+ sockWriter.forceLeave();
+ sockReader.forceStopRead();
- currSock = null;
+ currSock = null;
- queue.clear();
+ queue.clear();
- onDisconnected();
+ onDisconnected();
- UUID newId = UUID.randomUUID();
+ UUID newId = UUID.randomUUID();
- U.quietAndWarn(log, "Local node will try to
reconnect to cluster with new id due " +
- "to network problems [newId=" + newId +
- ", prevId=" + locNode.id() +
- ", locNode=" + locNode + ']');
+ U.quietAndWarn(log, "Local node will try to
reconnect to cluster with new id due " +
+ "to network problems [newId=" + newId +
+ ", prevId=" + locNode.id() +
+ ", locNode=" + locNode + ']');
- locNode.onClientDisconnected(newId);
+ locNode.onClientDisconnected(newId);
- throttleClientReconnect();
+ throttleClientReconnect();
- tryJoin();
- }
+ tryJoin();
}
- else if (msg instanceof TcpDiscoveryNodeFailedMessage
&&
-
((TcpDiscoveryNodeFailedMessage)msg).failedNodeId().equals(locNode.id())) {
- TcpDiscoveryNodeFailedMessage msg0 =
(TcpDiscoveryNodeFailedMessage)msg;
+ }
+ else if (msg instanceof TcpDiscoveryNodeFailedMessage &&
+
((TcpDiscoveryNodeFailedMessage)msg).failedNodeId().equals(locNode.id())) {
+ TcpDiscoveryNodeFailedMessage msg0 =
(TcpDiscoveryNodeFailedMessage)msg;
- assert msg0.force() : msg0;
+ assert msg0.force() : msg0;
- forceFailMsg = msg0;
- }
- else if (msg instanceof SocketClosedMessage) {
- if (((SocketClosedMessage)msg).sock == currSock) {
- Socket sock = currSock.sock;
+ forceFailMsg = msg0;
+ }
+ else if (msg instanceof SocketClosedMessage) {
+ if (((SocketClosedMessage)msg).sock == currSock) {
+ Socket sock = currSock.sock;
- InetSocketAddress prevAddr = new
InetSocketAddress(sock.getInetAddress(), sock.getPort());
+ InetSocketAddress prevAddr = new
InetSocketAddress(sock.getInetAddress(), sock.getPort());
- currSock = null;
+ currSock = null;
- boolean join = joinLatch.getCount() > 0;
+ boolean join = joinLatch.getCount() > 0;
- if (spi.getSpiContext().isStopping() || state
== SEGMENTED) {
- leaveLatch.countDown();
+ if (spi.getSpiContext().isStopping() || state ==
SEGMENTED) {
+ leaveLatch.countDown();
- if (join) {
- joinError(new
IgniteSpiException("Failed to connect to cluster: socket closed."));
+ if (join) {
+ joinError(new IgniteSpiException("Failed
to connect to cluster: socket closed."));
- break;
+ break;
+ }
+ }
+ else {
+ if (forceFailMsg != null) {
+ if (log.isDebugEnabled()) {
+ log.debug("Connection closed, local
node received force fail message, " +
+ "will not try to restore
connection");
}
+
+ queue.addFirst(SPI_RECONNECT_FAILED);
}
else {
- if (forceFailMsg != null) {
- if (log.isDebugEnabled()) {
- log.debug("Connection closed,
local node received force fail message, " +
- "will not try to restore
connection");
- }
-
- queue.addFirst(SPI_RECONNECT_FAILED);
- }
- else {
- if (log.isDebugEnabled())
- log.debug("Connection closed, will
try to restore connection.");
+ if (log.isDebugEnabled())
+ log.debug("Connection closed, will try
to restore connection.");
- assert reconnector == null;
+ assert reconnector == null;
- reconnector = new Reconnector(join,
prevAddr);
- reconnector.start();
- }
+ reconnector = new Reconnector(join,
prevAddr);
+ reconnector.start();
}
}
}
- else if (msg == SPI_RECONNECT_FAILED) {
- if (reconnector != null) {
- reconnector.cancel();
- reconnector.join();
+ }
+ else if (msg == SPI_RECONNECT_FAILED) {
+ if (reconnector != null) {
+ reconnector.cancel();
+ reconnector.join();
- reconnector = null;
- }
- else
- assert forceFailMsg != null;
-
- if (spi.isClientReconnectDisabled()) {
- if (state != SEGMENTED && state != STOPPED) {
- if (forceFailMsg != null) {
- U.quietAndWarn(log, "Local node was
dropped from cluster due to network problems " +
- "[nodeInitiatedFail=" +
forceFailMsg.creatorNodeId() +
- ", msg=" + forceFailMsg.warning()
+ ']');
- }
+ reconnector = null;
+ }
+ else
+ assert forceFailMsg != null;
- if (log.isDebugEnabled()) {
- log.debug("Failed to restore closed
connection, reconnect disabled, " +
- "local node segmented
[networkTimeout=" + spi.netTimeout + ']');
- }
+ if (spi.isClientReconnectDisabled()) {
+ if (state != SEGMENTED && state != STOPPED) {
+ if (forceFailMsg != null) {
+ U.quietAndWarn(log, "Local node was
dropped from cluster due to network problems " +
+ "[nodeInitiatedFail=" +
forceFailMsg.creatorNodeId() +
+ ", msg=" + forceFailMsg.warning() +
']');
+ }
- state = SEGMENTED;
+ if (log.isDebugEnabled()) {
+ log.debug("Failed to restore closed
connection, reconnect disabled, " +
+ "local node segmented
[networkTimeout=" + spi.netTimeout + ']');
+ }
- notifyDiscovery(
- EVT_NODE_SEGMENTED, topVer, locNode,
allVisibleNodes());
+ state = SEGMENTED;
+
+ notifyDiscovery(
+ EVT_NODE_SEGMENTED, topVer, locNode,
allVisibleNodes(), null);
+ }
+ }
+ else {
+ if (state == STARTING || state == CONNECTED) {
+ if (log.isDebugEnabled()) {
+ log.debug("Failed to restore closed
connection, will try to reconnect " +
+ "[networkTimeout=" + spi.netTimeout +
+ ", joinTimeout=" + spi.joinTimeout +
+ ", failMsg=" + forceFailMsg + ']');
}
+
+ onDisconnected();
}
- else {
- if (state == STARTING || state == CONNECTED) {
- if (log.isDebugEnabled()) {
- log.debug("Failed to restore closed
connection, will try to reconnect " +
- "[networkTimeout=" +
spi.netTimeout +
- ", joinTimeout=" + spi.joinTimeout
+
- ", failMsg=" + forceFailMsg + ']');
- }
- onDisconnected();
- }
+ UUID newId = UUID.randomUUID();
- UUID newId = UUID.randomUUID();
+ if (forceFailMsg != null) {
+ long delay =
IgniteSystemProperties.getLong(IGNITE_DISCO_FAILED_CLIENT_RECONNECT_DELAY,
+ DFLT_DISCO_FAILED_CLIENT_RECONNECT_DELAY);
- if (forceFailMsg != null) {
- long delay =
IgniteSystemProperties.getLong(IGNITE_DISCO_FAILED_CLIENT_RECONNECT_DELAY,
-
DFLT_DISCO_FAILED_CLIENT_RECONNECT_DELAY);
-
- if (delay > 0) {
- U.quietAndWarn(log, "Local node was
dropped from cluster due to network problems, " +
- "will try to reconnect with new id
after " + delay + "ms (reconnect delay " +
- "can be changed using
IGNITE_DISCO_FAILED_CLIENT_RECONNECT_DELAY system " +
- "property) [" +
- "newId=" + newId +
- ", prevId=" + locNode.id() +
- ", locNode=" + locNode +
- ", nodeInitiatedFail=" +
forceFailMsg.creatorNodeId() +
- ", msg=" + forceFailMsg.warning()
+ ']');
-
- Thread.sleep(delay);
- }
- else {
- U.quietAndWarn(log, "Local node was
dropped from cluster due to network problems, " +
- "will try to reconnect with new id
[" +
- "newId=" + newId +
- ", prevId=" + locNode.id() +
- ", locNode=" + locNode +
- ", nodeInitiatedFail=" +
forceFailMsg.creatorNodeId() +
- ", msg=" + forceFailMsg.warning()
+ ']');
- }
+ if (delay > 0) {
+ U.quietAndWarn(log, "Local node was
dropped from cluster due to network problems, " +
+ "will try to reconnect with new id
after " + delay + "ms (reconnect delay " +
+ "can be changed using
IGNITE_DISCO_FAILED_CLIENT_RECONNECT_DELAY system " +
+ "property) [" +
+ "newId=" + newId +
+ ", prevId=" + locNode.id() +
+ ", locNode=" + locNode +
+ ", nodeInitiatedFail=" +
forceFailMsg.creatorNodeId() +
+ ", msg=" + forceFailMsg.warning() +
']');
- forceFailMsg = null;
+ Thread.sleep(delay);
}
- else if (log.isInfoEnabled()) {
- log.info("Client node disconnected from
cluster, will try to reconnect with new id " +
- "[newId=" + newId + ", prevId=" +
locNode.id() + ", locNode=" + locNode + ']');
+ else {
+ U.quietAndWarn(log, "Local node was
dropped from cluster due to network problems, " +
+ "will try to reconnect with new id [" +
+ "newId=" + newId +
+ ", prevId=" + locNode.id() +
+ ", locNode=" + locNode +
+ ", nodeInitiatedFail=" +
forceFailMsg.creatorNodeId() +
+ ", msg=" + forceFailMsg.warning() +
']');
}
- locNode.onClientDisconnected(newId);
-
- tryJoin();
+ forceFailMsg = null;
+ }
+ else if (log.isInfoEnabled()) {
+ log.info("Client node disconnected from
cluster, will try to reconnect with new id " +
+ "[newId=" + newId + ", prevId=" +
locNode.id() + ", locNode=" + locNode + ']');
}
- }
- else {
- if (joining()) {
- IgniteSpiException err = null;
- if (dm instanceof
TcpDiscoveryDuplicateIdMessage)
- err =
spi.duplicateIdError((TcpDiscoveryDuplicateIdMessage)msg);
- else if (dm instanceof
TcpDiscoveryAuthFailedMessage)
- err =
spi.authenticationFailedError((TcpDiscoveryAuthFailedMessage)msg);
- //TODO:
https://issues.apache.org/jira/browse/IGNITE-9829
- else if (dm instanceof
TcpDiscoveryCheckFailedMessage)
- err =
spi.checkFailedError((TcpDiscoveryCheckFailedMessage)msg);
+ locNode.onClientDisconnected(newId);
- if (err != null) {
- if (state == DISCONNECTED) {
- U.error(log, "Failed to reconnect,
segment local node.", err);
+ tryJoin();
+ }
+ }
+ else {
+ TcpDiscoveryAbstractMessage discoMsg =
(TcpDiscoveryAbstractMessage)msg;
- state = SEGMENTED;
+ if (joining()) {
+ IgniteSpiException err = null;
- notifyDiscovery(
- EVT_NODE_SEGMENTED, topVer,
locNode, allVisibleNodes());
- }
- else
- joinError(err);
+ if (discoMsg instanceof
TcpDiscoveryDuplicateIdMessage)
+ err =
spi.duplicateIdError((TcpDiscoveryDuplicateIdMessage)msg);
+ else if (discoMsg instanceof
TcpDiscoveryAuthFailedMessage)
+ err =
spi.authenticationFailedError((TcpDiscoveryAuthFailedMessage)msg);
+ //TODO:
https://issues.apache.org/jira/browse/IGNITE-9829
+ else if (discoMsg instanceof
TcpDiscoveryCheckFailedMessage)
+ err =
spi.checkFailedError((TcpDiscoveryCheckFailedMessage)msg);
- cancel();
+ if (err != null) {
+ if (state == DISCONNECTED) {
+ U.error(log, "Failed to reconnect, segment
local node.", err);
- break;
+ state = SEGMENTED;
+
+ notifyDiscovery(
+ EVT_NODE_SEGMENTED, topVer, locNode,
allVisibleNodes(), null);
}
- }
+ else
+ joinError(err);
- processDiscoveryMessage(dm);
+ cancel();
+
+ break;
+ }
}
+
+ processDiscoveryMessage(discoMsg);
}
}
}
@@ -2107,10 +2103,15 @@ class ClientImpl extends TcpDiscoveryImpl {
sockReader.setSocket(joinRes, locNode.clientRouterNodeId());
}
- /**
- * @param msg Message.
- */
+ /** */
protected void processDiscoveryMessage(TcpDiscoveryAbstractMessage
msg) {
+ try (Scope ignored =
operationCtxDispatcher.restoreSnapshot(msg.opCtxSnp)) {
+ processDiscoveryMessage0(msg);
+ }
+ }
+
+ /** */
+ protected void processDiscoveryMessage0(TcpDiscoveryAbstractMessage
msg) {
assert msg != null;
assert msg.verified() || msg.senderNodeId() == null;
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 3a6ca0e6604..02a626abe55 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
@@ -2969,7 +2969,7 @@ class ServerImpl extends TcpDiscoveryImpl {
}
if (!fromSocket)
- msg.opCtxMsg =
operationCtxDispatcher.collectDistributedAttributeValues();
+
msg.attachOperationContextSnapshot(operationCtxDispatcher.createSnapshot());
boolean addFirst = msg.highPriority() && !ignoreHighPriority;
@@ -3198,7 +3198,7 @@ class ServerImpl extends TcpDiscoveryImpl {
if (msg == WAKEUP)
return;
- try (Scope ignored =
operationCtxDispatcher.restoreRemoteAttributeValues(msg.opCtxMsg)) {
+ try (Scope ignored =
operationCtxDispatcher.restoreSnapshot(msg.opCtxSnp)) {
processMessage0(msg);
}
}
@@ -5968,7 +5968,7 @@ class ServerImpl extends TcpDiscoveryImpl {
getLocalNodeId(), nextMsg);
ackMsg.topologyVersion(msg.topologyVersion());
- ackMsg.opCtxMsg =
operationCtxDispatcher.collectDistributedAttributeValues();
+
ackMsg.attachOperationContextSnapshot(operationCtxDispatcher.createSnapshot());
processCustomMessage(ackMsg, waitForNotification);
}
@@ -6012,10 +6012,6 @@ class ServerImpl extends TcpDiscoveryImpl {
synchronized (mux) {
joiningEmpty = joiningNodes.isEmpty();
-
- if (log.isDebugEnabled())
- log.debug("Delay custom message processing, there are
joining nodes [msg=" + msg +
- ", joiningNodes=" + joiningNodes + ']');
}
boolean delayMsg = msg.topologyVersion() == 0L && !joiningEmpty;
@@ -6025,6 +6021,10 @@ class ServerImpl extends TcpDiscoveryImpl {
pendingCustomMsgs.add(msg);
}
+ if (log.isDebugEnabled())
+ log.debug("Delay custom message processing, there are
joining nodes [msg=" + msg +
+ ", joiningNodes=" + joiningNodes + ']');
+
return true;
}
@@ -6095,8 +6095,11 @@ class ServerImpl extends TcpDiscoveryImpl {
if (joiningEmpty && isLocalNodeCoordinator()) {
TcpDiscoveryCustomEventMessage msg;
- while ((msg = pollPendingCustomMessage()) != null)
- processCustomMessage(msg, true);
+ while ((msg = pollPendingCustomMessage()) != null) {
+ try (Scope ignored =
operationCtxDispatcher.restoreSnapshot(msg.opCtxSnp)) {
+ processCustomMessage(msg, true);
+ }
+ }
}
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryAbstractMessage.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryAbstractMessage.java
index 25e67f80610..f6cb7562f86 100644
---
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryAbstractMessage.java
+++
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryAbstractMessage.java
@@ -22,7 +22,7 @@ import java.util.HashSet;
import java.util.Set;
import java.util.UUID;
import org.apache.ignite.internal.Order;
-import org.apache.ignite.internal.thread.context.OperationContextMessage;
+import
org.apache.ignite.internal.thread.context.OperationContextSnapshotMessage;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -43,6 +43,9 @@ public abstract class TcpDiscoveryAbstractMessage implements
Message {
/** */
protected static final int CLIENT_RECON_SUCCESS_FLAG_POS = 2;
+ /** */
+ protected static final int OP_CTX_ATTACHED_FLAG_POS = 3;
+
/** */
protected static final int FORCE_FAIL_FLAG_POS = 4;
@@ -77,10 +80,10 @@ public abstract class TcpDiscoveryAbstractMessage
implements Message {
@Order(4)
Set<UUID> failedNodes;
- /** Operation context attributes message. */
+ /** Operation context snapshot message. */
@GridToStringInclude
@Order(5)
- public @Nullable OperationContextMessage opCtxMsg;
+ public @Nullable OperationContextSnapshotMessage opCtxSnp;
/**
* Default no-arg constructor for {@link Externalizable} interface.
@@ -106,7 +109,7 @@ public abstract class TcpDiscoveryAbstractMessage
implements Message {
verifierNodeId = msg.verifierNodeId;
topVer = msg.topVer;
flags = msg.flags;
- opCtxMsg = msg.opCtxMsg;
+ opCtxSnp = msg.opCtxSnp;
}
/**
@@ -233,6 +236,15 @@ public abstract class TcpDiscoveryAbstractMessage
implements Message {
setFlag(FORCE_FAIL_FLAG_POS, force);
}
+ /** @param opCtxSnp Operation context snapshot. */
+ public void attachOperationContextSnapshot(@Nullable
OperationContextSnapshotMessage opCtxSnp) {
+ if (!getFlag(OP_CTX_ATTACHED_FLAG_POS)) {
+ this.opCtxSnp = opCtxSnp;
+
+ setFlag(OP_CTX_ATTACHED_FLAG_POS, true);
+ }
+ }
+
/**
* @param pos Flag position.
* @return Flag value.
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java
index 625f290d1c6..c3853aea01e 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java
@@ -710,7 +710,8 @@ public class DataStreamerImplSelfTest extends
GridCommonAbstractTest {
appMsg,
GridTestUtils.<Boolean>getFieldValue(ioMsg,
"ordered"),
ioMsg.timeout(),
- ioMsg.skipOnTimeout()
+ ioMsg.skipOnTimeout(),
+ null
);
needStaleTop = false;
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 17bcf34b58a..96ba2bdf6b3 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
@@ -17,7 +17,9 @@
package org.apache.ignite.internal.thread.context;
+import java.io.Serializable;
import java.util.Map;
+import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
@@ -28,6 +30,7 @@ import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.managers.communication.GridMessageListener;
import org.apache.ignite.internal.managers.communication.IgniteIoTestMessage;
import org.apache.ignite.internal.processors.authentication.User;
@@ -35,40 +38,65 @@ import
org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer;
import
org.apache.ignite.internal.processors.security.TestDiscoveryAcknowledgeMessage;
import org.apache.ignite.internal.processors.security.TestDiscoveryMessage;
import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.plugin.AbstractTestPluginProvider;
import org.apache.ignite.plugin.PluginContext;
import org.apache.ignite.spi.MessagesPluginProvider;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.ListeningTestLogger;
+import org.apache.ignite.testframework.LogListener;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.jetbrains.annotations.Nullable;
import org.junit.Test;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.ignite.internal.GridTopic.TOPIC_IO_TEST;
+import static
org.apache.ignite.internal.managers.communication.GridIoPolicy.SYSTEM_POOL;
import static
org.apache.ignite.internal.thread.context.OperationContextAttribute.newInstance;
import static
org.apache.ignite.internal.thread.context.OperationContextAttributePropagationTest.TestIgniteComponent.DFLT_PTR;
import static
org.apache.ignite.internal.thread.context.OperationContextAttributePropagationTest.TestIgniteComponent.DFLT_USR;
import static
org.apache.ignite.internal.thread.context.OperationContextAttributePropagationTest.TestIgniteComponent.PTR_ATTR;
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.testframework.GridTestUtils.assertThrows;
import static
org.apache.ignite.testframework.GridTestUtils.assertThrowsAnyCause;
import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+import static
org.apache.ignite.testframework.config.GridTestProperties.IGNITE_CFG_PREPROCESSOR_CLS;
+import static org.junit.Assume.assumeFalse;
/** */
public class OperationContextAttributePropagationTest extends
GridCommonAbstractTest {
+ /** */
+ public static final LogListener MSG_DELAYED_LSNR =
LogListener.builder().andMatches(
+ "Delay custom message processing, there are joining nodes"
+ ).build();
+
/** */
private volatile Consumer<Integer> discoMsgLsnr;
/** {@inheritDoc} */
- @Override protected void afterTest() throws Exception {
- super.afterTest();
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+ TestIgniteComponent.discoveryDataExchangeStartedLatch = null;
+ TestIgniteComponent.discoveryDataExchangeUnblockedLatch = null;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
stopAllGrids();
+
+ super.afterTest();
}
/** {@inheritDoc} */
@Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+ cfg.setGridLogger(new ListeningTestLogger(log, MSG_DELAYED_LSNR));
+
cfg.setPluginProviders(
new TestIgniteComponent(),
new MessagesPluginProvider(TestDiscoveryMessage.class,
TestDiscoveryAcknowledgeMessage.class));
@@ -84,6 +112,46 @@ public class OperationContextAttributePropagationTest
extends GridCommonAbstract
doTestOperationContextAttributesPropagationThroughDiscovery(new
WALPointer(1, 1, 1), User.create("1", "1"));
}
+ /** */
+ @Test
+ public void testPostponedDiscoveryMessage() throws Exception {
+ assumeFalse(System.getProperty(IGNITE_CFG_PREPROCESSOR_CLS,
"").contains("ZookeeperDiscoverySpi"));
+
+ prepareCluster();
+
+ WALPointer ptrVal = new WALPointer(1, 1, 1);
+ User usrVal = User.create("1", "1");
+
+ TestIgniteComponent.discoveryDataExchangeStartedLatch = new
CountDownLatch(1);
+ TestIgniteComponent.discoveryDataExchangeUnblockedLatch = new
CountDownLatch(1);
+
+ try {
+ IgniteInternalFuture<?> startFut = GridTestUtils.runAsync(() ->
startGrid(3));
+
+
assertTrue(discoveryDataExchangeStartedLatch.await(getTestTimeout(),
MILLISECONDS));
+
+ setLoggerDebugLevel();
+
+ MSG_DELAYED_LSNR.reset();
+
+ IgniteInternalFuture<?> checkFut = GridTestUtils.runAsync(() -> {
+ try (Scope ignored = OperationContext.set(PTR_ATTR, ptrVal,
USR_ATTR, usrVal)) {
+ checkOperationContextDiscoveryTransmission(0, ptrVal,
usrVal);
+ }
+ });
+
+ assertTrue(MSG_DELAYED_LSNR.check(getTestTimeout()));
+
+ discoveryDataExchangeUnblockedLatch.countDown();
+
+ startFut.get();
+ checkFut.get();
+ }
+ finally {
+ discoveryDataExchangeUnblockedLatch.countDown();
+ }
+ }
+
/** */
@Test
public void testSendAttributesByCommunication() throws Exception {
@@ -92,6 +160,62 @@ public class OperationContextAttributePropagationTest
extends GridCommonAbstract
doTestOperationContextAttributesPropagationThroughCommunication(new
WALPointer(1, 1, 1), User.create("1", "1"));
}
+ /** */
+ @Test
+ public void testPostponedCommunicationOrderedMessage() throws Exception {
+ prepareCluster();
+
+ for (int fromIdx = 0; fromIdx < 3; ++fromIdx) {
+ for (int toIdx = 0; toIdx < 3; ++toIdx) {
+ if (fromIdx == toIdx)
+ continue;
+
+ IgniteEx from = grid(fromIdx);
+ IgniteEx to = grid(toIdx);
+
+ CountDownLatch rcvLatch = new CountDownLatch(3);
+
+ GridMessageListener lsnr = (nodeId, msg, plc) -> {
+ if (msg instanceof User) {
+ assertEquals(msg, OperationContext.get(USR_ATTR));
+ assertEquals(DFLT_PTR, OperationContext.get(PTR_ATTR));
+
+ rcvLatch.countDown();
+ }
+ else if (msg instanceof WALPointer) {
+ assertEquals(msg, OperationContext.get(PTR_ATTR));
+ assertEquals(DFLT_USR, OperationContext.get(USR_ATTR));
+
+ rcvLatch.countDown();
+ }
+ };
+
+ User usrVal = User.create("1", "1");
+ WALPointer ptrVal = new WALPointer(2, 2, 2);
+
+ to.context().io().removeMessageListener(TOPIC_IO_TEST);
+ to.context().io().addMessageListener(TOPIC_IO_TEST, lsnr);
+
+ try {
+ try (Scope ignored = OperationContext.set(USR_ATTR,
usrVal)) {
+ from.context().io().sendOrderedMessage(node(from, to),
TOPIC_IO_TEST, usrVal, SYSTEM_POOL, 5_000, false);
+ }
+
+ from.context().io().sendOrderedMessage(node(from, to),
TOPIC_IO_TEST, DFLT_USR, SYSTEM_POOL, 5_000, false);
+
+ try (Scope ignored = OperationContext.set(PTR_ATTR,
ptrVal)) {
+ from.context().io().sendOrderedMessage(node(from, to),
TOPIC_IO_TEST, ptrVal, SYSTEM_POOL, 5_000, false);
+ }
+
+ assertTrue(rcvLatch.await(getTestTimeout(), MILLISECONDS));
+ }
+ finally {
+
assertTrue(to.context().io().removeMessageListener(TOPIC_IO_TEST, lsnr));
+ }
+ }
+ }
+ }
+
/** */
private void prepareCluster() throws Exception {
startGrids(2);
@@ -197,10 +321,11 @@ public class OperationContextAttributePropagationTest
extends GridCommonAbstract
IgniteEx from = grid(fromIdx);
IgniteEx to = grid(toIdx);
- CountDownLatch rcvLatch = new CountDownLatch(2);
+ // GridIoManager automatically sends response for IgniteIoTestMessage
+ CountDownLatch rcvLatch = new CountDownLatch(4);
GridMessageListener lsnr = (nodeId, msg, plc) -> {
- if (msg instanceof IgniteIoTestMessage &&
((IgniteIoTestMessage)msg).request()) {
+ if (msg instanceof IgniteIoTestMessage) {
assertEquals(expUsrVal, OperationContext.get(USR_ATTR));
assertEquals(expPtrVal, OperationContext.get(PTR_ATTR));
@@ -209,6 +334,7 @@ public class OperationContextAttributePropagationTest
extends GridCommonAbstract
};
to.context().io().addMessageListener(TOPIC_IO_TEST, lsnr);
+ from.context().io().addMessageListener(TOPIC_IO_TEST, lsnr);
try {
from.context().io().sendIoTest(node(from, to), null, false);
@@ -218,6 +344,7 @@ public class OperationContextAttributePropagationTest
extends GridCommonAbstract
}
finally {
assertTrue(to.context().io().removeMessageListener(TOPIC_IO_TEST,
lsnr));
+
assertTrue(from.context().io().removeMessageListener(TOPIC_IO_TEST, lsnr));
}
}
@@ -228,6 +355,12 @@ public class OperationContextAttributePropagationTest
extends GridCommonAbstract
/** */
static class TestIgniteComponent extends AbstractTestPluginProvider {
+ /** */
+ public static CountDownLatch discoveryDataExchangeUnblockedLatch;
+
+ /** */
+ public static CountDownLatch discoveryDataExchangeStartedLatch;
+
/** */
public static final WALPointer DFLT_PTR = new WALPointer(0, 0, 0);
@@ -240,14 +373,36 @@ public class OperationContextAttributePropagationTest
extends GridCommonAbstract
/** */
public static final OperationContextAttribute<User> USR_ATTR =
newInstance(DFLT_USR);
+ /** */
+ private GridKernalContext kctx;
+
/** {@inheritDoc} */
@Override public String name() {
return "TestDistributedOperationContextAttributesRegistrator";
}
+ /** {@inheritDoc} */
+ @Override public @Nullable Serializable provideDiscoveryData(UUID
nodeId) {
+ if (discoveryDataExchangeStartedLatch != null
+ && !kctx.localNodeId().equals(nodeId)
+ && !U.isLocalNodeCoordinator(kctx.discovery())
+ ) {
+ try {
+ discoveryDataExchangeStartedLatch.countDown();
+
+
assertTrue(discoveryDataExchangeUnblockedLatch.await(5_000, MILLISECONDS));
+ }
+ catch (InterruptedException e) {
+ throw new IgniteException(e);
+ }
+ }
+
+ return super.provideDiscoveryData(nodeId);
+ }
+
/** {@inheritDoc} */
@Override public void start(PluginContext ctx) {
- GridKernalContext kctx = ((IgniteEx)ctx.grid()).context();
+ kctx = ((IgniteEx)ctx.grid()).context();
kctx.operationContextDispatcher().registerDistributedAttribute(MAX_ATTRS_CNT -
1, USR_ATTR);
kctx.operationContextDispatcher().registerDistributedAttribute(0,
PTR_ATTR);
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 36940bfaa0a..c7a6437cb09 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
@@ -20,7 +20,7 @@ package org.apache.ignite.spi.discovery.zk.internal;
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.internal.thread.context.OperationContextMessage;
+import
org.apache.ignite.internal.thread.context.OperationContextSnapshotMessage;
import org.apache.ignite.plugin.extensions.communication.MessageFactory;
import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
@@ -30,7 +30,7 @@ import org.jetbrains.annotations.Nullable;
/**
* <p>A holder for effective attributes of distributed {@link
OperationContext}. Analogue of
- * {@link TcpDiscoveryAbstractMessage#opCtxMsg} while we do not use a common
base class for all the Zookeeper messages.</p>
+ * {@link TcpDiscoveryAbstractMessage#opCtxSnp} while we do not use a common
base class for all the Zookeeper messages.</p>
*
* <p>NOTE: The difference is also the limitation on message type. In {@link
TcpDiscoverySpi} we transfer distributed
* {@link OperationContext} with all the messages. In {@link
ZookeeperDiscoverySpi} with from-Ignite
@@ -46,7 +46,7 @@ public class ZkOperationContextAwareCustomMessage implements
DiscoverySpiCustomM
/** */
@Order(1)
- OperationContextMessage opCtxMsg;
+ OperationContextSnapshotMessage opCtxSnp;
/** Default constructor for {@link MessageFactory}. */
public ZkOperationContextAwareCustomMessage() {
@@ -55,21 +55,21 @@ public class ZkOperationContextAwareCustomMessage
implements DiscoverySpiCustomM
/**
* @param delegate Original message.
- * @param opCtxMsg Distributed operation context message.
+ * @param opCtxSnp Operation context snapshot message.
*/
- public ZkOperationContextAwareCustomMessage(DiscoverySpiCustomMessage
delegate, OperationContextMessage opCtxMsg) {
+ public ZkOperationContextAwareCustomMessage(DiscoverySpiCustomMessage
delegate, OperationContextSnapshotMessage opCtxSnp) {
assert delegate != null;
- assert opCtxMsg != null;
+ assert opCtxSnp != null;
assert !(delegate instanceof ZkOperationContextAwareCustomMessage);
this.delegate = delegate;
- this.opCtxMsg = opCtxMsg;
+ this.opCtxSnp = opCtxSnp;
}
/** {@inheritDoc} */
@Override public @Nullable DiscoverySpiCustomMessage ackMessage() {
DiscoverySpiCustomMessage ack = delegate.ackMessage();
- return ack == null ? null : new
ZkOperationContextAwareCustomMessage(ack, opCtxMsg);
+ return ack == null ? null : new
ZkOperationContextAwareCustomMessage(ack, opCtxSnp);
}
/** {@inheritDoc} */
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 ec8990c1cad..8e4ef420d42 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
@@ -67,7 +67,7 @@ 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.OperationContextMessage;
+import
org.apache.ignite.internal.thread.context.OperationContextSnapshotMessage;
import org.apache.ignite.internal.thread.context.Scope;
import org.apache.ignite.internal.thread.pool.IgniteThreadPoolExecutor;
import org.apache.ignite.internal.util.GridLongList;
@@ -667,10 +667,10 @@ public class ZookeeperDiscoveryImpl {
/** */
public void sendCustomEvent(DiscoverySpiCustomMessage msg) {
- OperationContextMessage opCtx =
opCtxDispatcher.collectDistributedAttributeValues();
+ OperationContextSnapshotMessage opCtxSnp =
opCtxDispatcher.createSnapshot();
- if (opCtx != null)
- sendCustomMessage(new ZkOperationContextAwareCustomMessage(msg,
opCtx));
+ if (opCtxSnp != null)
+ sendCustomMessage(new ZkOperationContextAwareCustomMessage(msg,
opCtxSnp));
else
sendCustomMessage(msg);
}
@@ -3523,10 +3523,10 @@ public class ZookeeperDiscoveryImpl {
* @param msg Custom message to process. Can be a {@link
ZkOperationContextAwareCustomMessage}.
*/
private void notifyCustomEvent(final ZkDiscoveryCustomEventData evtData,
DiscoverySpiCustomMessage msg) {
- OperationContextMessage opCtxMsg = null;
+ OperationContextSnapshotMessage opCtxSnp = null;
if (msg instanceof ZkOperationContextAwareCustomMessage) {
- opCtxMsg = ((ZkOperationContextAwareCustomMessage)msg).opCtxMsg;
+ opCtxSnp = ((ZkOperationContextAwareCustomMessage)msg).opCtxSnp;
msg = ((ZkOperationContextAwareCustomMessage)msg).delegate;
}
@@ -3543,7 +3543,7 @@ public class ZookeeperDiscoveryImpl {
IgniteFuture<?> fut;
- try (Scope ignored =
opCtxDispatcher.restoreRemoteAttributeValues(opCtxMsg)) {
+ try (Scope ignored = opCtxDispatcher.restoreSnapshot(opCtxSnp)) {
fut = lsnr.onDiscovery(
new DiscoveryNotification(
DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT,
