You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ti...@apache.org on 2024/12/16 16:08:51 UTC
(ignite) branch application_context updated: IGNITE-23723 Support SessionContext in CacheInterceptor (#11726)
This is an automated email from the ASF dual-hosted git repository.
timoninmaxim pushed a commit to branch application_context
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/application_context by this push:
new 107a1067359 IGNITE-23723 Support SessionContext in CacheInterceptor (#11726)
107a1067359 is described below
commit 107a106735976d9324023dae0714e9db22b383a6
Author: Maksim Timonin <ti...@apache.org>
AuthorDate: Mon Dec 16 19:08:43 2024 +0300
IGNITE-23723 Support SessionContext in CacheInterceptor (#11726)
---
.../src/main/java/org/apache/ignite/Ignite.java | 2 +
.../internal/IgniteApplicationAttributesAware.java | 5 +-
.../communication/GridIoMessageFactory.java | 4 +
.../processors/cache/GridCacheAdapter.java | 56 ++-
.../processors/cache/GridCacheContext.java | 89 ++--
.../processors/cache/GridCacheProcessor.java | 2 +-
.../GridDistributedTxPrepareRequest.java | 19 +
.../GridDistributedTxRemoteAdapter.java | 4 +
.../dht/TransactionAttributesAwareRequest.java | 167 +++++++
.../AtomicApplicationAttributesAwareRequest.java | 157 ++++++
.../distributed/dht/atomic/GridDhtAtomicCache.java | 38 +-
.../atomic/GridNearAtomicAbstractUpdateFuture.java | 48 +-
.../atomic/GridNearAtomicSingleUpdateFuture.java | 6 +-
.../dht/atomic/GridNearAtomicUpdateFuture.java | 16 +-
.../cache/transactions/IgniteInternalTx.java | 10 +
.../cache/transactions/IgniteTransactionsImpl.java | 30 +-
.../cache/transactions/IgniteTxAdapter.java | 13 +
.../cache/transactions/IgniteTxHandler.java | 18 +
.../cache/transactions/IgniteTxManager.java | 11 +-
.../internal/processors/odbc/ClientTxSupport.java | 3 +-
.../ignite/internal/util/GridArgumentCheck.java | 14 +
.../resources/SessionContextProviderResource.java | 3 +-
.../TxDataConsistencyOnCommitFailureTest.java | 2 +-
.../SessionContextCacheInterceptorTest.java | 535 +++++++++++++++++++++
.../testsuites/IgniteTaskSessionSelfTestSuite.java | 2 +
.../processors/cache/jta/CacheJtaManager.java | 1 +
26 files changed, 1180 insertions(+), 75 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/Ignite.java b/modules/core/src/main/java/org/apache/ignite/Ignite.java
index 3518066b600..135732756f5 100644
--- a/modules/core/src/main/java/org/apache/ignite/Ignite.java
+++ b/modules/core/src/main/java/org/apache/ignite/Ignite.java
@@ -22,6 +22,7 @@ import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import javax.cache.CacheException;
+import org.apache.ignite.cache.CacheInterceptor;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.affinity.Affinity;
import org.apache.ignite.cache.query.annotations.QuerySqlFunction;
@@ -798,6 +799,7 @@ public interface Ignite extends AutoCloseable {
* List of supported types of user defined functions that have access the attributes:
* <ul>
* <li>{@link QuerySqlFunction}</li>
+ * <li>{@link CacheInterceptor}</li>
* </ul>
*
* @param attrs Application attributes.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteApplicationAttributesAware.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteApplicationAttributesAware.java
index b0aaefc1199..265004cefe1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteApplicationAttributesAware.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteApplicationAttributesAware.java
@@ -58,6 +58,7 @@ import org.apache.ignite.configuration.CollectionConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.NearCacheConfiguration;
import org.apache.ignite.internal.processors.cache.IgniteCacheProxy;
+import org.apache.ignite.internal.processors.cache.transactions.IgniteTransactionsImpl;
import org.apache.ignite.internal.util.typedef.internal.A;
import org.apache.ignite.lang.IgniteProductVersion;
import org.apache.ignite.metric.IgniteMetrics;
@@ -299,7 +300,9 @@ public class IgniteApplicationAttributesAware implements Ignite {
/** {@inheritDoc} */
@Override public IgniteTransactions transactions() {
- return delegate.transactions();
+ IgniteTransactionsImpl<?, ?> txs = (IgniteTransactionsImpl<?, ?>)delegate.transactions();
+
+ return txs.withApplicationAttributes(attrs);
}
/** {@inheritDoc} */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
index c92a0b4fbac..729b302d45a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
@@ -73,6 +73,8 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrep
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareResponse;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtUnlockRequest;
import org.apache.ignite.internal.processors.cache.distributed.dht.PartitionUpdateCountersMessage;
+import org.apache.ignite.internal.processors.cache.distributed.dht.TransactionAttributesAwareRequest;
+import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.AtomicApplicationAttributesAwareRequest;
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicDeferredUpdateResponse;
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicNearResponse;
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicSingleUpdateRequest;
@@ -329,6 +331,8 @@ public class GridIoMessageFactory implements MessageFactoryProvider {
factory.register((short)177, TcpInverseConnectionResponseMessage::new);
factory.register(SnapshotFilesRequestMessage.TYPE_CODE, SnapshotFilesRequestMessage::new);
factory.register(SnapshotFilesFailureMessage.TYPE_CODE, SnapshotFilesFailureMessage::new);
+ factory.register((short)180, AtomicApplicationAttributesAwareRequest::new);
+ factory.register((short)181, TransactionAttributesAwareRequest::new);
// Incremental snapshot.
factory.register(IncrementalSnapshotAwareMessage.TYPE_CODE, IncrementalSnapshotAwareMessage::new);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
index b7cbb668a02..6b44b5e92f8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
@@ -1406,8 +1406,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
opCtx != null ? opCtx.readRepairStrategy() : null);
if (ctx.config().getInterceptor() != null)
- fut = fut.chain(new CX1<IgniteInternalFuture<V>, V>() {
- @Override public V applyx(IgniteInternalFuture<V> f) throws IgniteCheckedException {
+ fut = fut.chain(new CX1ContextAware<IgniteInternalFuture<V>, V>(opCtx) {
+ @Override public V apply0(IgniteInternalFuture<V> f) throws IgniteCheckedException {
K key = keepBinary ? (K)ctx.unwrapBinaryIfNeeded(key0, true, false, null) : key0;
return (V)ctx.config().getInterceptor().onGet(key, f.get());
@@ -1448,8 +1448,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
final boolean intercept = ctx.config().getInterceptor() != null;
IgniteInternalFuture<CacheEntry<K, V>> fr = fut.chain(
- new CX1<IgniteInternalFuture<EntryGetResult>, CacheEntry<K, V>>() {
- @Override public CacheEntry<K, V> applyx(IgniteInternalFuture<EntryGetResult> f)
+ new CX1ContextAware<IgniteInternalFuture<EntryGetResult>, CacheEntry<K, V>>(opCtx) {
+ @Override public CacheEntry<K, V> apply0(IgniteInternalFuture<EntryGetResult> f)
throws IgniteCheckedException {
EntryGetResult t = f.get();
@@ -1571,8 +1571,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
/*need ver*/false);
if (ctx.config().getInterceptor() != null)
- return fut.chain(new CX1<IgniteInternalFuture<Map<K, V>>, Map<K, V>>() {
- @Override public Map<K, V> applyx(IgniteInternalFuture<Map<K, V>> f) throws IgniteCheckedException {
+ return fut.chain(new CX1ContextAware<IgniteInternalFuture<Map<K, V>>, Map<K, V>>(opCtx) {
+ @Override public Map<K, V> apply0(IgniteInternalFuture<Map<K, V>> f) throws IgniteCheckedException {
return interceptGet(keys, f.get());
}
});
@@ -1615,8 +1615,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
final boolean intercept = ctx.config().getInterceptor() != null;
IgniteInternalFuture<Collection<CacheEntry<K, V>>> rf =
- fut.chain(new CX1<IgniteInternalFuture<Map<K, EntryGetResult>>, Collection<CacheEntry<K, V>>>() {
- @Override public Collection<CacheEntry<K, V>> applyx(
+ fut.chain(new CX1ContextAware<IgniteInternalFuture<Map<K, EntryGetResult>>, Collection<CacheEntry<K, V>>>(opCtx) {
+ @Override public Collection<CacheEntry<K, V>> apply0(
IgniteInternalFuture<Map<K, EntryGetResult>> f) throws IgniteCheckedException {
if (intercept)
return interceptGetEntries(keys, f.get());
@@ -3661,7 +3661,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
tCfg.getDefaultTxTimeout(),
!ctx.skipStore(),
0,
- null
+ null,
+ opCtx == null ? null : opCtx.applicationAttributes()
);
assert tx != null;
@@ -3781,7 +3782,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
txCfg.getDefaultTxTimeout(),
!skipStore,
0,
- null);
+ null,
+ opCtx == null ? null : opCtx.applicationAttributes());
return asyncOp(tx, op, opCtx, /*retry*/false);
}
@@ -4764,7 +4766,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
CU.transactionConfiguration(ctx, ctx.kernalContext().config()).getDefaultTxTimeout(),
opCtx == null || !opCtx.skipStore(),
0,
- null);
+ null,
+ opCtx == null ? null : opCtx.applicationAttributes());
IgniteInternalFuture<T> fut = asyncOp(tx, op, opCtx, retry);
@@ -6771,4 +6774,35 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
return entry != null && internalSet.contains(entry);
}
}
+
+ /** Wraps closure with CacheOperationContext. */
+ private abstract class CX1ContextAware<E1, R> extends CX1<E1, R> {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private final CacheOperationContext opCtx;
+
+ /** */
+ private CX1ContextAware(CacheOperationContext opCtx) {
+ this.opCtx = opCtx;
+ }
+
+ /** */
+ @Override public R applyx(E1 object) throws IgniteCheckedException {
+ CacheOperationContext prevOpCtx = ctx.operationContextPerCall();
+
+ ctx.operationContextPerCall(opCtx);
+
+ try {
+ return apply0(object);
+ }
+ finally {
+ ctx.operationContextPerCall(prevOpCtx);
+ }
+ }
+
+ /** */
+ abstract R apply0(E1 object) throws IgniteCheckedException;
+ }
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
index 5cfbadabb9b..ebdbfb6f8cd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
@@ -24,6 +24,7 @@ import java.io.InvalidObjectException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.io.ObjectStreamException;
+import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -44,6 +45,7 @@ import javax.cache.expiry.EternalExpiryPolicy;
import javax.cache.expiry.ExpiryPolicy;
import javax.cache.processor.EntryProcessorResult;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.binary.BinaryField;
@@ -62,6 +64,7 @@ import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.IgnitionEx;
import org.apache.ignite.internal.binary.BinaryMarshaller;
import org.apache.ignite.internal.binary.builder.BinaryObjectBuilderImpl;
+import org.apache.ignite.internal.cache.context.SessionContextImpl;
import org.apache.ignite.internal.managers.communication.GridIoManager;
import org.apache.ignite.internal.managers.deployment.GridDeploymentManager;
import org.apache.ignite.internal.managers.discovery.GridDiscoveryManager;
@@ -109,7 +112,6 @@ import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.GPC;
-import org.apache.ignite.internal.util.typedef.internal.GPR;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgnitePredicate;
@@ -117,6 +119,8 @@ import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.marshaller.Marshaller;
import org.apache.ignite.plugin.security.SecurityException;
import org.apache.ignite.plugin.security.SecurityPermission;
+import org.apache.ignite.session.SessionContext;
+import org.apache.ignite.session.SessionContextProvider;
import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_DISABLE_TRIGGERING_CACHE_INTERCEPTOR_ON_CONFLICT;
@@ -435,6 +439,14 @@ public class GridCacheContext<K, V> implements Externalizable {
assert locMacs != null;
}
+
+ try {
+ if (cacheCfg.getInterceptor() != null)
+ ctx.resource().injectToUdf(cacheCfg.getInterceptor(), new SessionContextProviderImpl(this));
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteException("Failed to inject resources to CacheInterceptor", e);
+ }
}
/**
@@ -1373,41 +1385,6 @@ public class GridCacheContext<K, V> implements Externalizable {
return isDht() || (isDhtAtomic() && dht().near() != null);
}
- /**
- * Creates Runnable that can be executed safely in a different thread inheriting
- * the same thread local projection as for the current thread. If no projection is
- * set for current thread then there's no need to create new object and method simply
- * returns given Runnable.
- *
- * @param r Runnable.
- * @return Runnable that can be executed in a different thread with the same
- * projection as for current thread.
- */
- public Runnable projectSafe(final Runnable r) {
- assert r != null;
-
- // Have to get operation context per call used by calling thread to use it in a new thread.
- final CacheOperationContext opCtx = operationContextPerCall();
-
- if (opCtx == null)
- return r;
-
- return new GPR() {
- @Override public void run() {
- CacheOperationContext old = operationContextPerCall();
-
- operationContextPerCall(opCtx);
-
- try {
- r.run();
- }
- finally {
- operationContextPerCall(old);
- }
- }
- };
- }
-
/**
* Creates callable that can be executed safely in a different thread inheriting
* the same thread local projection as for the current thread. If no projection is
@@ -2442,6 +2419,46 @@ public class GridCacheContext<K, V> implements Externalizable {
}
}
+ /** SessionContext provider to UDF. */
+ private static final class SessionContextProviderImpl implements SessionContextProvider, Serializable {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Emtpy session context. */
+ private static final SessionContext EMPTY = (attrName) -> null;
+
+ /** Cache context. */
+ private final GridCacheContext<?, ?> ctx;
+
+ /** */
+ SessionContextProviderImpl(GridCacheContext<?, ?> ctx) {
+ this.ctx = ctx;
+ }
+
+ /** {@inheritDoc} */
+ @Override public SessionContext getSessionContext() {
+ if (ctx.transactional()) {
+ IgniteInternalTx tx = ctx.cache().context().tm().tx();
+
+ if (tx != null) {
+ if (tx.applicationAttributes() == null)
+ return EMPTY;
+
+ return new SessionContextImpl(tx.applicationAttributes());
+ }
+ }
+
+ // It works for transactional caches also. For example, CacheInterceptor#onGet invoked out of the scope of
+ // a transaction context.
+ CacheOperationContext opCtx = ctx.operationContextPerCall();
+
+ if (opCtx == null || opCtx.applicationAttributes() == null)
+ return EMPTY;
+
+ return new SessionContextImpl(opCtx.applicationAttributes());
+ }
+ }
+
/** {@inheritDoc} */
@Override public String toString() {
return "GridCacheContext: " + name();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
index d91ef2f9d7d..f72d157fc60 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
@@ -599,7 +599,7 @@ public class GridCacheProcessor extends GridProcessorAdapter {
locCfgMgr = new GridLocalConfigManager(this, ctx);
- transactions = new IgniteTransactionsImpl(sharedCtx, null, false);
+ transactions = new IgniteTransactionsImpl(sharedCtx, null, false, null);
// Start shared managers.
for (GridCacheSharedManager mgr : sharedCtx.managers())
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxPrepareRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxPrepareRequest.java
index 73d704e0adb..783ecd0cf2a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxPrepareRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxPrepareRequest.java
@@ -149,6 +149,11 @@ public class GridDistributedTxPrepareRequest extends GridDistributedBaseMessage
@GridToStringExclude
private byte flags;
+ /** Application attributes. */
+ @GridDirectTransient
+ @GridToStringExclude
+ private @Nullable Map<String, String> appAttrs;
+
/**
* Required by {@link Externalizable}.
*/
@@ -362,6 +367,20 @@ public class GridDistributedTxPrepareRequest extends GridDistributedBaseMessage
return isFlag(LAST_REQ_FLAG_MASK);
}
+ /**
+ * @return Application attributes, or {@code null} if not set.
+ */
+ public Map<String, String> applicationAttributes() {
+ return appAttrs;
+ }
+
+ /**
+ * @param appAttrs Application attributes.
+ */
+ public void applicationAttributes(Map<String, String> appAttrs) {
+ this.appAttrs = appAttrs;
+ }
+
/** {@inheritDoc} */
@Override public IgniteTxState txState() {
return txState;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
index 9b5112af621..258bf5211cf 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
@@ -490,6 +490,8 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter imp
Set<GridDhtLocalPartition> reservedParts = new HashSet<>();
try {
+ cctx.tm().txContext(this);
+
Collection<IgniteTxEntry> entries = near() ? allEntries() : writeEntries();
// Data entry to write to WAL.
@@ -806,6 +808,8 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter imp
cctx.database().checkpointReadUnlock();
+ cctx.tm().resetContext();
+
if (wrapper != null)
wrapper.initialize(ret);
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/TransactionAttributesAwareRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/TransactionAttributesAwareRequest.java
new file mode 100644
index 00000000000..68b48b391f6
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/TransactionAttributesAwareRequest.java
@@ -0,0 +1,167 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.distributed.dht;
+
+import java.nio.ByteBuffer;
+import java.util.Map;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.GridDirectMap;
+import org.apache.ignite.internal.processors.cache.GridCacheMessage;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxPrepareRequest;
+import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
+import org.apache.ignite.plugin.extensions.communication.MessageReader;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+
+/** Wraps transaction prepare request with application attributes. */
+public class TransactionAttributesAwareRequest extends GridCacheMessage {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ public static final short TYPE_CODE = 181;
+
+ /** Original transaction prepare message. */
+ private GridDistributedTxPrepareRequest payload;
+
+ /** Application attributes. */
+ @GridDirectMap(keyType = String.class, valueType = String.class)
+ private Map<String, String> appAttrs;
+
+ /** */
+ public TransactionAttributesAwareRequest() {
+ }
+
+ /** */
+ public TransactionAttributesAwareRequest(
+ GridDistributedTxPrepareRequest payload,
+ Map<String, String> appAttrs
+ ) {
+ this.payload = payload;
+ this.appAttrs = appAttrs;
+ }
+
+ /** @return Original update message. */
+ public GridDistributedTxPrepareRequest payload() {
+ return payload;
+ }
+
+ /** @return Application attributes. */
+ public Map<String, String> applicationAttributes() {
+ return appAttrs;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void prepareMarshal(GridCacheSharedContext<?, ?> ctx) throws IgniteCheckedException {
+ payload.prepareMarshal(ctx);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void finishUnmarshal(GridCacheSharedContext<?, ?> ctx, ClassLoader ldr) throws IgniteCheckedException {
+ payload.finishUnmarshal(ctx, ldr);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
+ writer.setBuffer(buf);
+
+ if (!super.writeTo(buf, writer))
+ return false;
+
+ if (!writer.isHeaderWritten()) {
+ if (!writer.writeHeader(directType(), fieldsCount()))
+ return false;
+
+ writer.onHeaderWritten();
+ }
+
+ switch (writer.state()) {
+ case 3:
+ if (!writer.writeMap("appAttrs", appAttrs, MessageCollectionItemType.STRING, MessageCollectionItemType.STRING))
+ return false;
+
+ writer.incrementState();
+
+ case 4:
+ if (!writer.writeMessage("payload", payload))
+ return false;
+
+ writer.incrementState();
+
+ }
+
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
+ reader.setBuffer(buf);
+
+ if (!reader.beforeMessageRead())
+ return false;
+
+ if (!super.readFrom(buf, reader))
+ return false;
+
+ switch (reader.state()) {
+ case 3:
+ appAttrs = reader.readMap("appAttrs", MessageCollectionItemType.STRING, MessageCollectionItemType.STRING, false);
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 4:
+ payload = reader.readMessage("payload");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ }
+
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte fieldsCount() {
+ return 5;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int handlerId() {
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean cacheGroupMessage() {
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean addDeploymentInfo() {
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return TYPE_CODE;
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/AtomicApplicationAttributesAwareRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/AtomicApplicationAttributesAwareRequest.java
new file mode 100644
index 00000000000..7df4910490f
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/AtomicApplicationAttributesAwareRequest.java
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.distributed.dht.atomic;
+
+import java.nio.ByteBuffer;
+import java.util.Map;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.GridDirectMap;
+import org.apache.ignite.internal.processors.cache.GridCacheIdMessage;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
+import org.apache.ignite.plugin.extensions.communication.MessageReader;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+
+/** Wraps atomic updates with application attributes. */
+public class AtomicApplicationAttributesAwareRequest extends GridCacheIdMessage {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ public static final short TYPE_CODE = 180;
+
+ /** Original update message. */
+ private GridNearAtomicAbstractUpdateRequest payload;
+
+ /** Application attributes. */
+ @GridDirectMap(keyType = String.class, valueType = String.class)
+ private Map<String, String> appAttrs;
+
+ /** */
+ public AtomicApplicationAttributesAwareRequest() {
+ }
+
+ /** */
+ public AtomicApplicationAttributesAwareRequest(
+ GridNearAtomicAbstractUpdateRequest payload,
+ Map<String, String> appAttrs
+ ) {
+ this.payload = payload;
+ this.appAttrs = appAttrs;
+ cacheId = payload.cacheId();
+ }
+
+ /** @return Original update message. */
+ public GridNearAtomicAbstractUpdateRequest payload() {
+ return payload;
+ }
+
+ /** @return Application attributes. */
+ public Map<String, String> applicationAttributes() {
+ return appAttrs;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void prepareMarshal(GridCacheSharedContext<?, ?> ctx) throws IgniteCheckedException {
+ payload.prepareMarshal(ctx);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void finishUnmarshal(GridCacheSharedContext<?, ?> ctx, ClassLoader ldr) throws IgniteCheckedException {
+ payload.finishUnmarshal(ctx, ldr);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
+ writer.setBuffer(buf);
+
+ if (!super.writeTo(buf, writer))
+ return false;
+
+ if (!writer.isHeaderWritten()) {
+ if (!writer.writeHeader(directType(), fieldsCount()))
+ return false;
+
+ writer.onHeaderWritten();
+ }
+
+ switch (writer.state()) {
+ case 4:
+ if (!writer.writeMap("appAttrs", appAttrs, MessageCollectionItemType.STRING, MessageCollectionItemType.STRING))
+ return false;
+
+ writer.incrementState();
+
+ case 5:
+ if (!writer.writeMessage("payload", payload))
+ return false;
+
+ writer.incrementState();
+
+ }
+
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
+ reader.setBuffer(buf);
+
+ if (!reader.beforeMessageRead())
+ return false;
+
+ if (!super.readFrom(buf, reader))
+ return false;
+
+ switch (reader.state()) {
+ case 4:
+ appAttrs = reader.readMap("appAttrs", MessageCollectionItemType.STRING, MessageCollectionItemType.STRING, false);
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 5:
+ payload = reader.readMessage("payload");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ }
+
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte fieldsCount() {
+ return 6;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean addDeploymentInfo() {
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return TYPE_CODE;
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index 95774ba9480..984b688956d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -278,6 +278,27 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
});
+ ctx.io().addCacheHandler(
+ ctx.cacheId(),
+ ctx.startTopologyVersion(),
+ AtomicApplicationAttributesAwareRequest.class,
+ new CI2<UUID, AtomicApplicationAttributesAwareRequest>() {
+ @Override public void apply(
+ UUID nodeId,
+ AtomicApplicationAttributesAwareRequest req
+ ) {
+ if (req.applicationAttributes() != null)
+ ctx.operationContextPerCall(new CacheOperationContext().setApplicationAttributes(req.applicationAttributes()));
+
+ try {
+ processNearAtomicUpdateRequest(nodeId, req.payload());
+ }
+ finally {
+ ctx.operationContextPerCall(null);
+ }
+ }
+ });
+
ctx.io().addCacheHandler(
ctx.cacheId(),
ctx.startTopologyVersion(),
@@ -1089,7 +1110,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
opCtx != null && opCtx.skipStore(),
opCtx != null && opCtx.isKeepBinary(),
opCtx != null && opCtx.recovery(),
- opCtx != null && opCtx.noRetries() ? 1 : MAX_RETRIES);
+ opCtx != null && opCtx.noRetries() ? 1 : MAX_RETRIES,
+ opCtx != null ? opCtx.applicationAttributes() : null);
if (async) {
return asyncOp(new CO<IgniteInternalFuture<Object>>() {
@@ -1275,7 +1297,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
opCtx != null && opCtx.skipStore(),
opCtx != null && opCtx.isKeepBinary(),
opCtx != null && opCtx.recovery(),
- opCtx != null && opCtx.noRetries() ? 1 : MAX_RETRIES
+ opCtx != null && opCtx.noRetries() ? 1 : MAX_RETRIES,
+ opCtx != null ? opCtx.applicationAttributes() : null
);
}
else {
@@ -1296,7 +1319,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
opCtx != null && opCtx.skipStore(),
opCtx != null && opCtx.isKeepBinary(),
opCtx != null && opCtx.recovery(),
- opCtx != null && opCtx.noRetries() ? 1 : MAX_RETRIES);
+ opCtx != null && opCtx.noRetries() ? 1 : MAX_RETRIES,
+ opCtx != null ? opCtx.applicationAttributes() : null);
}
}
@@ -1353,7 +1377,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
opCtx != null && opCtx.skipStore(),
opCtx != null && opCtx.isKeepBinary(),
opCtx != null && opCtx.recovery(),
- opCtx != null && opCtx.noRetries() ? 1 : MAX_RETRIES);
+ opCtx != null && opCtx.noRetries() ? 1 : MAX_RETRIES,
+ opCtx != null ? opCtx.applicationAttributes() : null);
if (async) {
return asyncOp(new CO<IgniteInternalFuture<Object>>() {
@@ -3127,6 +3152,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
drPutVals = null;
}
+ CacheOperationContext opCtx = ctx.operationContextPerCall();
+
GridNearAtomicUpdateFuture updateFut = new GridNearAtomicUpdateFuture(
ctx,
this,
@@ -3144,7 +3171,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
req.skipStore(),
req.keepBinary(),
req.recovery(),
- MAX_RETRIES);
+ MAX_RETRIES,
+ opCtx == null ? null : opCtx.applicationAttributes());
updateFut.map();
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicAbstractUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicAbstractUpdateFuture.java
index 001b6927830..7c8fef96e66 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicAbstractUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicAbstractUpdateFuture.java
@@ -37,10 +37,12 @@ import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheEntryPredicate;
+import org.apache.ignite.internal.processors.cache.CacheOperationContext;
import org.apache.ignite.internal.processors.cache.CachePartialUpdateCheckedException;
import org.apache.ignite.internal.processors.cache.GridCacheAtomicFuture;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheFutureAdapter;
+import org.apache.ignite.internal.processors.cache.GridCacheMessage;
import org.apache.ignite.internal.processors.cache.GridCacheMvccManager;
import org.apache.ignite.internal.processors.cache.GridCacheOperation;
import org.apache.ignite.internal.processors.cache.GridCacheReturn;
@@ -109,6 +111,9 @@ public abstract class GridNearAtomicAbstractUpdateFuture extends GridCacheFuture
/** Recovery flag. */
protected final boolean recovery;
+ /** Application attributes. */
+ protected final @Nullable Map<String, String> appAttrs;
+
/** Near cache flag. */
protected final boolean nearEnabled;
@@ -158,6 +163,7 @@ public abstract class GridNearAtomicAbstractUpdateFuture extends GridCacheFuture
* @param keepBinary Keep binary flag.
* @param recovery {@code True} if cache operation is called in recovery mode.
* @param remapCnt Remap count.
+ * @param appAttrs Application attributes.
*/
protected GridNearAtomicAbstractUpdateFuture(
GridCacheContext cctx,
@@ -172,7 +178,8 @@ public abstract class GridNearAtomicAbstractUpdateFuture extends GridCacheFuture
boolean skipStore,
boolean keepBinary,
boolean recovery,
- int remapCnt
+ int remapCnt,
+ @Nullable Map<String, String> appAttrs
) {
if (log == null) {
msgLog = cctx.shared().atomicMessageLogger();
@@ -196,6 +203,7 @@ public abstract class GridNearAtomicAbstractUpdateFuture extends GridCacheFuture
nearEnabled = CU.isNearEnabled(cctx);
this.remapCnt = remapCnt;
+ this.appAttrs = appAttrs;
}
/**
@@ -291,17 +299,18 @@ public abstract class GridNearAtomicAbstractUpdateFuture extends GridCacheFuture
*/
final void sendSingleRequest(UUID nodeId, GridNearAtomicAbstractUpdateRequest req) {
if (cctx.localNodeId().equals(nodeId)) {
- cache.updateAllAsyncInternal(cctx.localNode(), req,
- (ignored, res) -> {
+ cache.updateAllAsyncInternal(cctx.localNode(), req, new UpdateReplyClosureContextAware() {
+ @Override void apply0(GridNearAtomicAbstractUpdateRequest req, GridNearAtomicUpdateResponse res) {
if (syncMode != FULL_ASYNC)
onPrimaryResponse(res.nodeId(), res, false);
else if (res.remapTopologyVersion() != null)
((GridDhtAtomicCache<?, ?>)cctx.cache()).remapToNewPrimary(req);
- });
+ }
+ });
}
else {
try {
- cctx.io().send(req.nodeId(), req, cctx.ioPolicy());
+ cctx.io().send(req.nodeId(), wrapWithApplicationAttributes(req), cctx.ioPolicy());
if (msgLog.isDebugEnabled()) {
msgLog.debug("Near update fut, sent request [futId=" + req.futureId() +
@@ -893,6 +902,35 @@ public abstract class GridNearAtomicAbstractUpdateFuture extends GridCacheFuture
ALL_RCVD_CHECK_PRIMARY
}
+ /** Wraps closure with CacheOperationContext awared of application attributes. */
+ protected abstract class UpdateReplyClosureContextAware implements GridDhtAtomicCache.UpdateReplyClosure {
+ /** */
+ @Override public void apply(GridNearAtomicAbstractUpdateRequest req, GridNearAtomicUpdateResponse res) {
+ CacheOperationContext prevOpCtx = cctx.operationContextPerCall();
+
+ if (appAttrs != null)
+ cctx.operationContextPerCall(new CacheOperationContext().setApplicationAttributes(appAttrs));
+
+ try {
+ apply0(req, res);
+ }
+ finally {
+ cctx.operationContextPerCall(prevOpCtx);
+ }
+ }
+
+ /** */
+ abstract void apply0(GridNearAtomicAbstractUpdateRequest req, GridNearAtomicUpdateResponse res);
+ }
+
+ /**
+ * @param msg Message to wrap if application attributes are specified.
+ * @return Original or wrapped message.
+ */
+ protected GridCacheMessage wrapWithApplicationAttributes(GridNearAtomicAbstractUpdateRequest msg) {
+ return appAttrs != null ? new AtomicApplicationAttributesAwareRequest(msg, appAttrs) : msg;
+ }
+
/** {@inheritDoc} */
@Override public String toString() {
return S.toString(GridNearAtomicAbstractUpdateFuture.class, this, super.toString());
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicSingleUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicSingleUpdateFuture.java
index db5af0d82c3..2322b9eeb4d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicSingleUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicSingleUpdateFuture.java
@@ -97,7 +97,8 @@ public class GridNearAtomicSingleUpdateFuture extends GridNearAtomicAbstractUpda
boolean skipStore,
boolean keepBinary,
boolean recovery,
- int remapCnt
+ int remapCnt,
+ @Nullable Map<String, String> appAttrs
) {
super(cctx,
cache,
@@ -111,7 +112,8 @@ public class GridNearAtomicSingleUpdateFuture extends GridNearAtomicAbstractUpda
skipStore,
keepBinary,
recovery,
- remapCnt);
+ remapCnt,
+ appAttrs);
this.key = key;
this.val = val;
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
index 08e221c2fad..426e64c1311 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
@@ -110,6 +110,7 @@ public class GridNearAtomicUpdateFuture extends GridNearAtomicAbstractUpdateFutu
* @param skipStore Skip store flag.
* @param keepBinary Keep binary flag.
* @param remapCnt Maximum number of retries.
+ * @param appAttrs Application attributes.
*/
public GridNearAtomicUpdateFuture(
GridCacheContext cctx,
@@ -128,7 +129,8 @@ public class GridNearAtomicUpdateFuture extends GridNearAtomicAbstractUpdateFutu
boolean skipStore,
boolean keepBinary,
boolean recovery,
- int remapCnt
+ int remapCnt,
+ @Nullable Map<String, String> appAttrs
) {
super(
cctx,
@@ -143,7 +145,8 @@ public class GridNearAtomicUpdateFuture extends GridNearAtomicAbstractUpdateFutu
skipStore,
keepBinary,
recovery,
- remapCnt);
+ remapCnt,
+ appAttrs);
assert vals == null || vals.size() == keys.size();
assert conflictPutVals == null || conflictPutVals.size() == keys.size();
@@ -668,7 +671,7 @@ public class GridNearAtomicUpdateFuture extends GridNearAtomicAbstractUpdateFutu
if (req.initMappingLocally() && reqState.mappedNodes.isEmpty())
reqState.resetLocalMapping();
- cctx.io().send(req.nodeId(), req, cctx.ioPolicy());
+ cctx.io().send(req.nodeId(), wrapWithApplicationAttributes(req), cctx.ioPolicy());
if (msgLog.isDebugEnabled()) {
msgLog.debug("Near update fut, sent request [futId=" + req.futureId() +
@@ -688,13 +691,14 @@ public class GridNearAtomicUpdateFuture extends GridNearAtomicAbstractUpdateFutu
}
if (locUpdate != null) {
- cache.updateAllAsyncInternal(cctx.localNode(), locUpdate,
- (req, res) -> {
+ cache.updateAllAsyncInternal(cctx.localNode(), locUpdate, new UpdateReplyClosureContextAware() {
+ @Override void apply0(GridNearAtomicAbstractUpdateRequest req, GridNearAtomicUpdateResponse res) {
if (syncMode != FULL_ASYNC)
onPrimaryResponse(res.nodeId(), res, false);
else if (res.remapTopologyVersion() != null)
((GridDhtAtomicCache<?, ?>)cctx.cache()).remapToNewPrimary(req);
- });
+ }
+ });
}
if (syncMode == FULL_ASYNC)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java
index 80a4eb7c3f0..92764327fa9 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java
@@ -664,6 +664,16 @@ public interface IgniteInternalTx {
*/
public void incrementalSnapshotId(@Nullable UUID id);
+ /**
+ * @return Application attributes.
+ */
+ public @Nullable Map<String, String> applicationAttributes();
+
+ /**
+ * @param appAttrs Application attributes.
+ */
+ public void applicationAttributes(Map<String, String> appAttrs);
+
/**
* @return Transaction counters.
* @param createIfAbsent {@code True} if non-null instance is needed.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTransactionsImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTransactionsImpl.java
index f3ad8b85cb4..95c3c2acba6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTransactionsImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTransactionsImpl.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.processors.cache.transactions;
import java.util.Collection;
+import java.util.Map;
import org.apache.ignite.IgniteTransactions;
import org.apache.ignite.configuration.TransactionConfiguration;
import org.apache.ignite.internal.IgniteTransactionsEx;
@@ -53,14 +54,23 @@ public class IgniteTransactionsImpl<K, V> implements IgniteTransactionsEx {
/** Tracing enabled flag. */
private boolean tracingEnabled;
+ /** Application attributes. */
+ private Map<String, String> appAttrs;
+
/**
* @param cctx Cache shared context.
* @param lb Label.
*/
- public IgniteTransactionsImpl(GridCacheSharedContext<K, V> cctx, @Nullable String lb, boolean tracingEnabled) {
+ public IgniteTransactionsImpl(
+ GridCacheSharedContext<K, V> cctx,
+ @Nullable String lb,
+ boolean tracingEnabled,
+ @Nullable Map<String, String> appAttrs
+ ) {
this.cctx = cctx;
this.lb = lb;
this.tracingEnabled = tracingEnabled;
+ this.appAttrs = appAttrs;
}
/** {@inheritDoc} */
@@ -193,7 +203,8 @@ public class IgniteTransactionsImpl<K, V> implements IgniteTransactionsEx {
timeout,
true,
txSize,
- lb
+ lb,
+ appAttrs
);
assert tx != null;
@@ -238,12 +249,23 @@ public class IgniteTransactionsImpl<K, V> implements IgniteTransactionsEx {
@Override public IgniteTransactions withLabel(String lb) {
A.notNull(lb, "label should not be empty.");
- return new IgniteTransactionsImpl<>(cctx, lb, tracingEnabled);
+ return new IgniteTransactionsImpl<>(cctx, lb, tracingEnabled, appAttrs);
}
/** {@inheritDoc} */
@Override public IgniteTransactions withTracing() {
- return new IgniteTransactionsImpl<>(cctx, lb, true);
+ return new IgniteTransactionsImpl<>(cctx, lb, true, appAttrs);
+ }
+
+ /**
+ * Returns an instance of {@code IgniteTransactions} with application attributes.
+ *
+ * @return Application attributes aware instance.
+ */
+ public IgniteTransactions withApplicationAttributes(Map<String, String> appAttrs) {
+ A.notEmpty(appAttrs, "application attributes");
+
+ return new IgniteTransactionsImpl<>(cctx, lb, tracingEnabled, appAttrs);
}
/**
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
index 0a36d340880..ce8c022e85c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
@@ -274,6 +274,9 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter implement
/** Transaction from which this transaction was copied by(if it was). */
private GridNearTxLocal parentTx;
+ /** Application attributes. */
+ private @Nullable Map<String, String> appAttrs;
+
/**
* @param cctx Cache registry.
* @param xidVer Transaction ID.
@@ -408,6 +411,16 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter implement
incSnpId = id;
}
+ /** {@inheritDoc} */
+ @Override public @Nullable Map<String, String> applicationAttributes() {
+ return appAttrs;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void applicationAttributes(Map<String, String> appAttrs) {
+ this.appAttrs = appAttrs;
+ }
+
/**
* @param parentTx Transaction from which this transaction was copied by.
*/
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
index 2224173cc92..8278caea4a9 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
@@ -47,6 +47,7 @@ import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.distributed.GridCacheTxRecoveryFuture;
import org.apache.ignite.internal.processors.cache.distributed.GridCacheTxRecoveryRequest;
import org.apache.ignite.internal.processors.cache.distributed.GridCacheTxRecoveryResponse;
+import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxPrepareRequest;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxRemoteAdapter;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxFinishFuture;
@@ -59,6 +60,7 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrep
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareResponse;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxRemote;
import org.apache.ignite.internal.processors.cache.distributed.dht.PartitionUpdateCountersMessage;
+import org.apache.ignite.internal.processors.cache.distributed.dht.TransactionAttributesAwareRequest;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtInvalidPartitionException;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopology;
@@ -232,6 +234,19 @@ public class IgniteTxHandler {
ctx.io().addCacheHandler(GridCacheTxRecoveryResponse.class, this::processCheckPreparedTxResponse);
ctx.io().addCacheHandler(IncrementalSnapshotAwareMessage.class, this::processIncrementalSnapshotAwareMessage);
+
+ ctx.io().addCacheHandler(TransactionAttributesAwareRequest.class, this::processAttributesAwareMessage);
+ }
+
+ /** */
+ private void processAttributesAwareMessage(UUID nodeId, TransactionAttributesAwareRequest msg) {
+ GridDistributedTxPrepareRequest txMsg = msg.payload();
+
+ txMsg.applicationAttributes(msg.applicationAttributes());
+
+ ctx.io()
+ .cacheHandler(COMMON_MESSAGE_HANDLER_ID, txMsg.getClass())
+ .apply(nodeId, txMsg);
}
/** */
@@ -584,6 +599,7 @@ public class IgniteTxHandler {
tx.explicitLock(true);
tx.transactionNodes(req.transactionNodes());
+ tx.applicationAttributes(req.applicationAttributes());
if (req.near())
tx.nearOnOriginatingNode(true);
@@ -1682,6 +1698,7 @@ public class IgniteTxHandler {
tx.onePhaseCommit(req.onePhaseCommit());
tx.writeVersion(req.writeVersion());
+ tx.applicationAttributes(req.applicationAttributes());
tx = ctx.tm().onCreated(null, tx);
@@ -1909,6 +1926,7 @@ public class IgniteTxHandler {
);
tx.writeVersion(req.writeVersion());
+ tx.applicationAttributes(req.applicationAttributes());
if (!tx.empty()) {
tx = ctx.tm().onCreated(null, tx);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
index 40c3881ef9d..e9cd1228322 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
@@ -78,9 +78,11 @@ import org.apache.ignite.internal.processors.cache.distributed.GridCacheMappedVe
import org.apache.ignite.internal.processors.cache.distributed.GridCacheTxRecoveryFuture;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedCacheEntry;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedLockCancelledException;
+import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxPrepareRequest;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxLocal;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxOnePhaseCommitAckRequest;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxRemote;
+import org.apache.ignite.internal.processors.cache.distributed.dht.TransactionAttributesAwareRequest;
import org.apache.ignite.internal.processors.cache.distributed.dht.colocated.GridDhtColocatedLockFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtInvalidPartitionException;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheAdapter;
@@ -674,6 +676,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
* @param timeout transaction timeout.
* @param txSize Expected transaction size.
* @param lb Label.
+ * @param appAttrs Application attributes.
* @return New transaction.
*/
public GridNearTxLocal newTx(
@@ -685,7 +688,8 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
long timeout,
boolean storeEnabled,
int txSize,
- @Nullable String lb
+ @Nullable String lb,
+ @Nullable Map<String, String> appAttrs
) {
assert sysCacheCtx == null || sysCacheCtx.systemTx();
@@ -716,6 +720,8 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
tx.topologyVersion(topVer);
}
+ tx.applicationAttributes(appAttrs);
+
return onCreated(sysCacheCtx, tx);
}
@@ -2953,6 +2959,9 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
* @param plc IO policy.
*/
public void sendTransactionMessage(UUID nodeId, GridCacheMessage msg, IgniteInternalTx tx, byte plc) throws IgniteCheckedException {
+ if (msg instanceof GridDistributedTxPrepareRequest && tx != null && tx.applicationAttributes() != null)
+ msg = new TransactionAttributesAwareRequest((GridDistributedTxPrepareRequest)msg, tx.applicationAttributes());
+
BiFunction<GridCacheMessage, IgniteInternalTx, GridCacheMessage> transform = txMsgTransform;
if (transform != null)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientTxSupport.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientTxSupport.java
index df3939679e6..79d8e2228d3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientTxSupport.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientTxSupport.java
@@ -59,7 +59,8 @@ public interface ClientTxSupport {
timeout,
true,
0,
- lb
+ lb,
+ null
);
}
finally {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/GridArgumentCheck.java b/modules/core/src/main/java/org/apache/ignite/internal/util/GridArgumentCheck.java
index 9b2704c29b7..2644595bc82 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/GridArgumentCheck.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/GridArgumentCheck.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.util;
import java.util.Collection;
+import java.util.Map;
import org.jetbrains.annotations.Nullable;
@@ -123,6 +124,19 @@ public class GridArgumentCheck {
throw new IllegalArgumentException(INVALID_ARG_MSG_PREFIX + name + NOT_EMPTY_SUFFIX);
}
+ /**
+ * Checks that given map is not empty.
+ *
+ * @param map Map.
+ * @param name Argument name.
+ */
+ public static void notEmpty(Map<?, ?> map, String name) {
+ notNull(map, name);
+
+ if (map.isEmpty())
+ throw new IllegalArgumentException(INVALID_ARG_MSG_PREFIX + name + NOT_EMPTY_SUFFIX);
+ }
+
/**
* Checks that given array is not empty.
*
diff --git a/modules/core/src/main/java/org/apache/ignite/resources/SessionContextProviderResource.java b/modules/core/src/main/java/org/apache/ignite/resources/SessionContextProviderResource.java
index e5921794ef3..bda80f30b2d 100644
--- a/modules/core/src/main/java/org/apache/ignite/resources/SessionContextProviderResource.java
+++ b/modules/core/src/main/java/org/apache/ignite/resources/SessionContextProviderResource.java
@@ -22,6 +22,7 @@ import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
+import org.apache.ignite.cache.CacheInterceptor;
import org.apache.ignite.cache.query.annotations.QuerySqlFunction;
import org.apache.ignite.session.SessionContext;
@@ -43,7 +44,7 @@ import org.apache.ignite.session.SessionContext;
* }
* </pre>
* <p>
- * Note, only {@link QuerySqlFunction} in the Calcite engine is supported.
+ * Note, {@link CacheInterceptor}, {@link QuerySqlFunction} in the Calcite engine are supported.
*
* @see SessionContext
*/
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxDataConsistencyOnCommitFailureTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxDataConsistencyOnCommitFailureTest.java
index 6699d96f218..d610090451e 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxDataConsistencyOnCommitFailureTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxDataConsistencyOnCommitFailureTest.java
@@ -187,7 +187,7 @@ public class TxDataConsistencyOnCommitFailureTest extends GridCommonAbstractTest
}
}).when(mockTm).
newTx(locTx.implicit(), locTx.implicitSingle(), null, locTx.concurrency(),
- locTx.isolation(), locTx.timeout(), locTx.storeEnabled(), locTx.size(), locTx.label());
+ locTx.isolation(), locTx.timeout(), locTx.storeEnabled(), locTx.size(), locTx.label(), locTx.applicationAttributes());
ctx.setTxManager(mockTm);
}
diff --git a/modules/core/src/test/java/org/apache/ignite/session/SessionContextCacheInterceptorTest.java b/modules/core/src/test/java/org/apache/ignite/session/SessionContextCacheInterceptorTest.java
new file mode 100644
index 00000000000..de31c237bbc
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/session/SessionContextCacheInterceptorTest.java
@@ -0,0 +1,535 @@
+/*
+ * 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.session;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import javax.cache.Cache;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheEntry;
+import org.apache.ignite.cache.CacheInterceptor;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteInterruptedCheckedException;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.apache.ignite.resources.SessionContextProviderResource;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+import org.jetbrains.annotations.Nullable;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.junit.Assume.assumeFalse;
+import static org.junit.Assume.assumeTrue;
+
+/** */
+@RunWith(Parameterized.class)
+public class SessionContextCacheInterceptorTest extends GridCommonAbstractTest {
+ /** */
+ private static final int KEYS = 10;
+
+ /** */
+ @Parameterized.Parameter
+ public CacheAtomicityMode mode;
+
+ /** */
+ @Parameterized.Parameter(1)
+ public CacheWriteSynchronizationMode syncMode;
+
+ /** */
+ @Parameterized.Parameter(2)
+ public boolean clnNode;
+
+ /** */
+ @Parameterized.Parameter(3)
+ public boolean dynamicCache;
+
+ /** */
+ @Parameterized.Parameter(4)
+ public int backups;
+
+ /** */
+ private Ignite ign;
+
+ /** */
+ private IgniteCache<Integer, String> cache;
+
+ /** */
+ @Parameterized.Parameters(name = "mode={0}, syncMode={1}, clnNode={2}, dynamicCache={3}, backups={4}")
+ public static Collection<Object[]> params() {
+ List<Object[]> params = new ArrayList<>();
+
+ for (CacheAtomicityMode m: CacheAtomicityMode.values()) {
+ for (CacheWriteSynchronizationMode syncMode : CacheWriteSynchronizationMode.values()) {
+ for (boolean cln : new boolean[] {true, false}) {
+ for (boolean dynCache : new boolean[] {true, false}) {
+ if (m == CacheAtomicityMode.TRANSACTIONAL) {
+ for (int backupCnt = 0; backupCnt <= 2; backupCnt++)
+ params.add(new Object[] {m, syncMode, cln, dynCache, backupCnt});
+ }
+ else
+ params.add(new Object[] {m, syncMode, cln, dynCache, 1});
+ }
+ }
+ }
+ }
+
+ return params;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ Collection<Object[]> params = params();
+
+ List<CacheConfiguration<?, ?>> ccfgs = new ArrayList<>();
+
+ for (Object[] param : params) {
+ boolean cln = (boolean)param[2];
+ boolean dyn = (boolean)param[3];
+
+ if (!dyn && !cln) {
+ ccfgs.add(
+ cacheConfig((CacheAtomicityMode)param[0], (CacheWriteSynchronizationMode)param[1], (int)param[4], false, false)
+ );
+ }
+ }
+
+ cfg.setCacheConfiguration(ccfgs.toArray(new CacheConfiguration<?, ?>[0]));
+
+ return cfg;
+ }
+
+ /** */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ startGrids(3);
+
+ startClientGrid(3);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ ign = grid(0);
+
+ if (clnNode)
+ ign = grid(3);
+
+ cache = ign.getOrCreateCache(cacheConfig(mode, syncMode, backups, dynamicCache, clnNode));
+
+ cache.clear();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ stopAllGrids();
+ }
+
+ /** */
+ @Test
+ public void testGetOperations() {
+ // CacheInterceptor#onGet doesn't work for client nodes.
+ assumeFalse("https://issues.apache.org/jira/browse/IGNITE-23810", clnNode);
+
+ for (int i = 0; i < KEYS; i++)
+ cache.put(i, String.valueOf(i));
+
+ IgniteCache<Integer, String> cacheApp = ign
+ .withApplicationAttributes(F.asMap("onGet", "sessionOnGet"))
+ .cache(cacheName());
+
+ for (int i = 0; i < KEYS; i++) {
+ final int j = i;
+
+ assertEquals("sessionOnGet" + j, () -> cacheApp.get(j));
+ assertEquals("sessionOnGet" + j, () -> cacheApp.getEntry(j).getValue());
+
+ assertEquals("sessionOnGet" + j, () -> cacheApp.getAsync(j).get(getTestTimeout()));
+ assertEquals("sessionOnGet" + j, () -> cacheApp.getEntryAsync(j).get(getTestTimeout()).getValue());
+
+ assertEquals(String.valueOf(j), () -> cache.get(j));
+ }
+
+ Set<Integer> keys = IntStream.range(0, KEYS).boxed().collect(Collectors.toSet());
+
+ Map<Integer, String> getAll = cacheApp.getAll(keys);
+ Map<Integer, String> getAllAsync = cacheApp.getAllAsync(keys).get(getTestTimeout());
+
+ for (int i = 0; i < KEYS; i++) {
+ final int j = i;
+
+ assertEquals("sessionOnGet" + j, () -> getAll.get(j));
+ assertEquals("sessionOnGet" + j, () -> getAllAsync.get(j));
+ }
+
+ Collection<CacheEntry<Integer, String>> entries = cacheApp.getEntries(keys);
+ Collection<CacheEntry<Integer, String>> entriesAsync = cacheApp.getEntriesAsync(keys).get(getTestTimeout());
+
+ for (CacheEntry<Integer, String> entry: entries)
+ assertTrue(entry.getValue().contains("sessionOnGet"));
+
+ for (CacheEntry<Integer, String> entry: entriesAsync)
+ assertTrue(entry.getValue().contains("sessionOnGet"));
+ }
+
+ /** */
+ @Test
+ public void testGetAndPut() {
+ IgniteCache<Integer, String> cacheApp = ign
+ .withApplicationAttributes(F.asMap("onBeforePut", "sessionOnPut"))
+ .cache(cacheName());
+
+ for (int i = 0; i < KEYS; i++) {
+ cacheApp.getAndPut(i, String.valueOf(i));
+ cacheApp.getAndPutIfAbsent(KEYS + i, String.valueOf(i));
+
+ cacheApp.getAndPutAsync(2 * KEYS + i, String.valueOf(i)).get(getTestTimeout());
+ cacheApp.getAndPutIfAbsentAsync(3 * KEYS + i, String.valueOf(i)).get(getTestTimeout());
+ }
+
+ for (int i = 0; i < 4 * KEYS; i++) {
+ final int j = i;
+
+ assertEquals("sessionOnPut" + j, () -> cacheApp.get(j));
+ }
+ }
+
+ /** */
+ @Test
+ public void testGetAndReplace() {
+ for (int i = 0; i < 2 * KEYS; i++)
+ cache.put(i, String.valueOf(i));
+
+ IgniteCache<Integer, String> cacheApp = ign
+ .withApplicationAttributes(F.asMap("onBeforePut", "sessionOnPut"))
+ .cache(cacheName());
+
+ for (int i = 0; i < KEYS; i++) {
+ final int j = i;
+
+ // Wait while FULL_ASYNC put finishes.
+ assertEquals(String.valueOf(j), () -> cache.get(j));
+ assertEquals(String.valueOf(KEYS + j), () -> cache.get(KEYS + j));
+
+ String ret = cacheApp.getAndReplace(i, "0" + i);
+ String retAsync = cacheApp.getAndReplaceAsync(KEYS + i, "1" + i).get(getTestTimeout());
+
+ if (syncMode != CacheWriteSynchronizationMode.FULL_ASYNC) {
+ assertEquals(String.valueOf(i), ret);
+ assertEquals(String.valueOf(KEYS + i), retAsync);
+ }
+
+ assertEquals("sessionOnPut" + j, () -> cacheApp.get(j));
+ assertEquals("sessionOnPut" + (KEYS + j), () -> cacheApp.get(KEYS + j));
+ }
+ }
+
+ /** */
+ @Test
+ public void testPutOperations() {
+ IgniteCache<Integer, String> cacheApp = ign
+ .withApplicationAttributes(F.asMap("onBeforePut", "sessionOnPut"))
+ .cache(cacheName());
+
+ for (int i = 0; i < KEYS; i++) {
+ cacheApp.put(i, String.valueOf(i));
+ cacheApp.putAsync(KEYS + i, String.valueOf(i)).get(getTestTimeout());
+
+ cacheApp.putIfAbsent(KEYS * 2 + i, String.valueOf(i));
+ cacheApp.putIfAbsentAsync(KEYS * 3 + i, String.valueOf(i));
+ }
+
+ Map<Integer, String> all = IntStream.range(KEYS * 4, KEYS * 5)
+ .boxed()
+ .collect(Collectors.toMap(k -> k, String::valueOf));
+
+ Map<Integer, String> allAsync = IntStream.range(KEYS * 5, KEYS * 6)
+ .boxed()
+ .collect(Collectors.toMap(k -> k, String::valueOf));
+
+ cacheApp.putAll(all);
+ cacheApp.putAllAsync(allAsync).get(getTestTimeout());
+
+ for (int i = 0; i < KEYS * 6; i++) {
+ final int j = i;
+
+ assertEquals("sessionOnPut" + j, () -> cache.get(j));
+ }
+ }
+
+ /** */
+ @Test
+ public void testOnBeforeRemove() {
+ for (int i = 0; i < KEYS; i++)
+ cache.put(i, String.valueOf(i));
+
+ IgniteCache<Integer, String> cacheApp = ign
+ .withApplicationAttributes(F.asMap("onBeforeRemove", "sessionOnRemove"))
+ .cache(cacheName());
+
+ for (int i = 0; i < KEYS; i++) {
+ final int j = i;
+
+ // Wait while FULL_ASYNC put finishes.
+ assertEquals(String.valueOf(j), () -> cache.get(j));
+
+ boolean removed = cacheApp.remove(i);
+
+ // Transaction actually return other value. Because the interception performs at commit time.
+ // FULL_ASYNC mode doesn't care about return value.
+ if (mode != CacheAtomicityMode.TRANSACTIONAL && syncMode != CacheWriteSynchronizationMode.FULL_ASYNC)
+ assertFalse(removed);
+
+ cacheApp.remove(i);
+ assertEquals(String.valueOf(j), () -> cache.get(j));
+ }
+
+ Set<Integer> keys = IntStream.range(0, KEYS).boxed().collect(Collectors.toSet());
+ cacheApp.removeAll(keys);
+ cacheApp.removeAllAsync(keys).get(getTestTimeout());
+
+ for (int i = 0; i < KEYS; i++) {
+ final int j = i;
+
+ assertEquals(String.valueOf(j), () -> cache.get(j));
+ }
+ }
+
+ /** */
+ @Test
+ public void testMultiCacheTransaction() throws Exception {
+ assumeTrue(mode == CacheAtomicityMode.TRANSACTIONAL);
+
+ ign.createCache(new CacheConfiguration<Integer, String>()
+ .setName(cacheName("new-cache"))
+ .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
+ .setInterceptor(new SessionContextCacheInterceptor()));
+
+ awaitPartitionMapExchange();
+
+ Ignite ignApp = ign
+ .withApplicationAttributes(F.asMap("onBeforePut", "sessionOnPut"));
+
+ int mult = 0;
+
+ for (TransactionConcurrency txConc: TransactionConcurrency.values()) {
+ for (TransactionIsolation txIsol: TransactionIsolation.values()) {
+ for (boolean async: new boolean[] {false, true}) {
+ try (Transaction tx = ignApp.transactions().txStart(txConc, txIsol)) {
+ for (int i = mult * KEYS; i < (mult + 1) * KEYS; i++) {
+ ignApp.cache(cacheName()).put(i, String.valueOf(i));
+ ignApp.cache(cacheName("new-cache")).put(i, String.valueOf(i));
+ }
+
+ if (async)
+ tx.commitAsync().get(getTestTimeout());
+ else
+ tx.commit();
+
+ mult++;
+ }
+ }
+ }
+ }
+
+ for (int i = 0; i < mult * KEYS; i++) {
+ final int j = i;
+
+ assertEquals("sessionOnPut" + j, () -> cache.get(j));
+ assertEquals("sessionOnPut" + j, () -> ign.cache(cacheName("new-cache")).get(j));
+ }
+ }
+
+ /** */
+ @Test
+ public void testExplicitSingleKeyTransaction() {
+ assumeTrue(mode == CacheAtomicityMode.TRANSACTIONAL);
+
+ Ignite ignApp = ign
+ .withApplicationAttributes(F.asMap("onBeforePut", "sessionOnPut"));
+
+ int key = 0;
+
+ for (TransactionConcurrency txConc: TransactionConcurrency.values()) {
+ for (TransactionIsolation txIsol: TransactionIsolation.values()) {
+ for (boolean async: new boolean[] {false, true}) {
+ try (Transaction tx = ignApp.transactions().txStart(txConc, txIsol)) {
+ ignApp.cache(cacheName()).put(key, String.valueOf(key));
+
+ if (async)
+ tx.commitAsync().get(getTestTimeout());
+ else
+ tx.commit();
+
+ key++;
+ }
+ }
+ }
+ }
+
+ for (int i = 0; i < key; i++) {
+ final int j = i;
+
+ assertEquals("sessionOnPut" + j, () -> cache.get(j));
+ }
+ }
+
+ /** */
+ @Test
+ public void testMultipleApplicationAttributes() {
+ assumeFalse("https://issues.apache.org/jira/browse/IGNITE-23810", clnNode);
+
+ IgniteCache<Integer, String> cacheApp = ign
+ .withApplicationAttributes(F.asMap(
+ "onGet", "sessionOnGet",
+ "onBeforePut", "sessionOnPut"))
+ .cache(cacheName());
+
+ cache.put(0, "0");
+
+ assertEquals("sessionOnGet" + 0, () -> cacheApp.get(0));
+
+ cacheApp.put(1, "1");
+
+ assertEquals("sessionOnPut" + 1, () -> cache.get(1));
+ }
+
+ /** */
+ @Test
+ public void testWithNoRetries() {
+ IgniteCache<Object, Object> cacheApp = ign
+ .withApplicationAttributes(F.asMap("onBeforePut", "sessionOnPut"))
+ .cache(cacheName())
+ .withNoRetries();
+
+ for (int i = 0; i < KEYS; i++)
+ cacheApp.putAsync((i), String.valueOf(i));
+
+ for (int i = 0; i < KEYS; i++) {
+ final int j = i;
+
+ assertEquals("sessionOnPut" + j, () -> cache.get(j));
+ }
+ }
+
+ /** */
+ private void assertEquals(Object exp, Supplier<Object> act) {
+ if (syncMode == CacheWriteSynchronizationMode.FULL_SYNC)
+ assertEquals(exp, act.get());
+ else {
+ try {
+ assertTrue(GridTestUtils.waitForCondition(() -> exp.equals(act.get()), getTestTimeout(), 10L));
+ }
+ catch (IgniteInterruptedCheckedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ /** */
+ private static final class SessionContextCacheInterceptor implements CacheInterceptor<Integer, String> {
+ /** */
+ @SessionContextProviderResource
+ private SessionContextProvider sesCtxPrv;
+
+ /** */
+ @Override public @Nullable String onGet(Integer key, @Nullable String val) {
+ String ret = sesCtxPrv.getSessionContext().getAttribute("onGet");
+
+ return ret == null ? val : ret + key;
+ }
+
+ /** */
+ @Override public @Nullable String onBeforePut(Cache.Entry<Integer, String> entry, String newVal) {
+ String ret = sesCtxPrv.getSessionContext().getAttribute("onBeforePut");
+
+ return ret == null ? newVal : ret + entry.getKey();
+ }
+
+ /** */
+ @Override public void onAfterPut(Cache.Entry<Integer, String> entry) {
+
+ }
+
+ /** */
+ @Override public @Nullable IgniteBiTuple<Boolean, String> onBeforeRemove(Cache.Entry<Integer, String> entry) {
+ String ret = sesCtxPrv.getSessionContext().getAttribute("onBeforeRemove");;
+
+ return new IgniteBiTuple<>(ret != null, entry.getValue());
+ }
+
+ /** */
+ @Override public void onAfterRemove(Cache.Entry<Integer, String> entry) {
+
+ }
+ }
+
+ /** */
+ private String cacheName() {
+ return cacheName(DEFAULT_CACHE_NAME);
+ }
+
+ /** */
+ private String cacheName(String baseName) {
+ return cacheName(baseName, mode, syncMode, backups, dynamicCache, clnNode);
+ }
+
+ /** */
+ private String cacheName(
+ String baseName,
+ CacheAtomicityMode mode,
+ CacheWriteSynchronizationMode syncMode,
+ int backups,
+ boolean dynamicCache,
+ boolean clnNode
+ ) {
+ return baseName + "_" + mode + "_" + syncMode + "_" + backups + "_" + dynamicCache + "_" + clnNode;
+ }
+
+ /** */
+ private CacheConfiguration<Integer, String> cacheConfig(
+ CacheAtomicityMode mode,
+ CacheWriteSynchronizationMode syncMode,
+ int backups,
+ boolean dynamicCache,
+ boolean clnNode
+ ) {
+ return new CacheConfiguration<Integer, String>()
+ .setName(cacheName(DEFAULT_CACHE_NAME, mode, syncMode, backups, dynamicCache, clnNode))
+ .setAtomicityMode(mode)
+ .setWriteSynchronizationMode(syncMode)
+ .setBackups(backups)
+ .setInterceptor(new SessionContextCacheInterceptor());
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteTaskSessionSelfTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteTaskSessionSelfTestSuite.java
index db60e8f4340..9c958c94365 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteTaskSessionSelfTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteTaskSessionSelfTestSuite.java
@@ -37,6 +37,7 @@ import org.apache.ignite.session.GridSessionSetTaskAttributeSelfTest;
import org.apache.ignite.session.GridSessionTaskWaitJobAttributeSelfTest;
import org.apache.ignite.session.GridSessionWaitAttributeSelfTest;
import org.apache.ignite.session.IgniteWithApplicationAttributesAwareTest;
+import org.apache.ignite.session.SessionContextCacheInterceptorTest;
import org.junit.runner.RunWith;
import org.junit.runners.Suite;
@@ -65,6 +66,7 @@ import org.junit.runners.Suite;
GridSessionCollisionSpiSelfTest.class,
GridSessionCheckpointSelfTest.class,
IgniteWithApplicationAttributesAwareTest.class,
+ SessionContextCacheInterceptorTest.class,
})
public class IgniteTaskSessionSelfTestSuite {
}
diff --git a/modules/jta/src/main/java/org/apache/ignite/internal/processors/cache/jta/CacheJtaManager.java b/modules/jta/src/main/java/org/apache/ignite/internal/processors/cache/jta/CacheJtaManager.java
index 8dcdd574fa6..b98a7f8b006 100644
--- a/modules/jta/src/main/java/org/apache/ignite/internal/processors/cache/jta/CacheJtaManager.java
+++ b/modules/jta/src/main/java/org/apache/ignite/internal/processors/cache/jta/CacheJtaManager.java
@@ -168,6 +168,7 @@ public class CacheJtaManager extends CacheJtaManagerAdapter {
tCfg.getDefaultTxTimeout(),
/*store enabled*/true,
/*tx size*/0,
+ null,
null
);
}