This is an automated email from the ASF dual-hosted git repository.
anton-vinogradov 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 32bbb11c1e5 IGNITE-28936 Remove dead marshalling code from
GridDhtAtomicUpdateRequest (#13419)
32bbb11c1e5 is described below
commit 32bbb11c1e52c4c1ec4be981521495fce8f4e299
Author: Anton Vinogradov <[email protected]>
AuthorDate: Fri Jul 31 16:15:16 2026 +0300
IGNITE-28936 Remove dead marshalling code from GridDhtAtomicUpdateRequest
(#13419)
---
.../ignite/internal/CoreMessagesProvider.java | 2 +-
.../atomic/GridDhtAtomicAbstractUpdateFuture.java | 7 -
.../atomic/GridDhtAtomicAbstractUpdateRequest.java | 27 ----
.../distributed/dht/atomic/GridDhtAtomicCache.java | 39 +-----
.../atomic/GridDhtAtomicSingleUpdateFuture.java | 2 -
.../atomic/GridDhtAtomicSingleUpdateRequest.java | 29 ----
.../dht/atomic/GridDhtAtomicUpdateFuture.java | 2 -
.../dht/atomic/GridDhtAtomicUpdateRequest.java | 149 +--------------------
.../distributed/near/GridNearAtomicCache.java | 14 +-
9 files changed, 18 insertions(+), 253 deletions(-)
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java
b/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java
index fa6514e2148..ffabd975489 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
@@ -521,7 +521,7 @@ public class CoreMessagesProvider extends
AbstractMarshallableMessageFactoryProv
withSchema(GridDhtForceKeysRequest.class);
withSchema(GridDhtForceKeysResponse.class);
withNoSchema(GridDhtAtomicDeferredUpdateResponse.class);
- withSchema(GridDhtAtomicUpdateRequest.class);
+ withNoSchema(GridDhtAtomicUpdateRequest.class);
withSchema(GridDhtAtomicUpdateResponse.class);
withSchema(GridNearAtomicFullUpdateRequest.class);
withSchema(GridDhtAtomicSingleUpdateRequest.class);
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateFuture.java
index 612d01dc178..05326e9d67f 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateFuture.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateFuture.java
@@ -23,7 +23,6 @@ import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;
-import javax.cache.processor.EntryProcessor;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cluster.ClusterNode;
@@ -132,7 +131,6 @@ public abstract class GridDhtAtomicAbstractUpdateFuture
extends GridCacheFutureA
* @param affAssignment Affinity assignment.
* @param entry Entry to map.
* @param val Value to write.
- * @param entryProc Entry processor.
* @param ttl TTL (optional).
* @param conflictExpireTime Conflict expire time (optional).
* @param conflictVer Conflict version (optional).
@@ -147,7 +145,6 @@ public abstract class GridDhtAtomicAbstractUpdateFuture
extends GridCacheFutureA
AffinityAssignment affAssignment,
GridDhtCacheEntry entry,
@Nullable CacheObject val,
- EntryProcessor<Object, Object, Object> entryProc,
long ttl,
long conflictExpireTime,
@Nullable GridCacheVersion conflictVer,
@@ -196,7 +193,6 @@ public abstract class GridDhtAtomicAbstractUpdateFuture
extends GridCacheFutureA
updateReq.addWriteValue(entry.key(),
val,
- entryProc,
ttl,
conflictExpireTime,
conflictVer,
@@ -225,7 +221,6 @@ public abstract class GridDhtAtomicAbstractUpdateFuture
extends GridCacheFutureA
* @param readers Entry readers.
* @param entry Entry.
* @param val Value.
- * @param entryProc Entry processor..
* @param ttl TTL for near cache update (optional).
* @param expireTime Expire time for near cache update (optional).
* @param readRepairRecovery Recovery on Read Repair.
@@ -235,7 +230,6 @@ public abstract class GridDhtAtomicAbstractUpdateFuture
extends GridCacheFutureA
GridDhtCacheEntry.ReaderId[] readers,
GridDhtCacheEntry entry,
@Nullable CacheObject val,
- EntryProcessor<Object, Object, Object> entryProc,
long ttl,
long expireTime,
boolean readRepairRecovery) {
@@ -285,7 +279,6 @@ public abstract class GridDhtAtomicAbstractUpdateFuture
extends GridCacheFutureA
updateReq.addNearWriteValue(entry.key(),
val,
- entryProc,
ttl,
expireTime);
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateRequest.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateRequest.java
index aa9639c449e..1684f3ce48d 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateRequest.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateRequest.java
@@ -18,7 +18,6 @@
package org.apache.ignite.internal.processors.cache.distributed.dht.atomic;
import java.util.UUID;
-import javax.cache.processor.EntryProcessor;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.Order;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
@@ -249,11 +248,6 @@ public abstract class GridDhtAtomicAbstractUpdateRequest
extends GridCacheIdMess
return addDepInfo;
}
- /**
- * @return Force transform backups flag.
- */
- public abstract boolean forceTransformBackups();
-
/** {@inheritDoc} */
@Override public IgniteLogger messageLogger(GridCacheSharedContext<?, ?>
ctx) {
return ctx.atomicMessageLogger();
@@ -262,7 +256,6 @@ public abstract class GridDhtAtomicAbstractUpdateRequest
extends GridCacheIdMess
/**
* @param key Key to add.
* @param val Value, {@code null} if should be removed.
- * @param entryProc Entry processor.
* @param ttl TTL (optional).
* @param conflictExpireTime Conflict expire time (optional).
* @param conflictVer Conflict version (optional).
@@ -273,7 +266,6 @@ public abstract class GridDhtAtomicAbstractUpdateRequest
extends GridCacheIdMess
*/
public abstract void addWriteValue(KeyCacheObject key,
@Nullable CacheObject val,
- EntryProcessor<Object, Object, Object> entryProc,
long ttl,
long conflictExpireTime,
@Nullable GridCacheVersion conflictVer,
@@ -285,13 +277,11 @@ public abstract class GridDhtAtomicAbstractUpdateRequest
extends GridCacheIdMess
/**
* @param key Key to add.
* @param val Value, {@code null} if should be removed.
- * @param entryProc Entry processor.
* @param ttl TTL.
* @param expireTime Expire time.
*/
public abstract void addNearWriteValue(KeyCacheObject key,
@Nullable CacheObject val,
- EntryProcessor<Object, Object, Object> entryProc,
long ttl,
long expireTime);
@@ -379,24 +369,12 @@ public abstract class GridDhtAtomicAbstractUpdateRequest
extends GridCacheIdMess
*/
@Nullable public abstract CacheObject previousValue(int idx);
- /**
- * @param idx Key index.
- * @return Entry processor.
- */
- @Nullable public abstract EntryProcessor<Object, Object, Object>
entryProcessor(int idx);
-
/**
* @param idx Near key index.
* @return Value.
*/
@Nullable public abstract CacheObject nearValue(int idx);
- /**
- * @param idx Key index.
- * @return Transform closure.
- */
- @Nullable public abstract EntryProcessor<Object, Object, Object>
nearEntryProcessor(int idx);
-
/**
* @param idx Index.
* @return Conflict version.
@@ -427,11 +405,6 @@ public abstract class GridDhtAtomicAbstractUpdateRequest
extends GridCacheIdMess
*/
public abstract long nearExpireTime(int idx);
- /**
- * @return Optional arguments for entry processor.
- */
- @Nullable public abstract Object[] invokeArguments();
-
/**
* Sets flag mask.
*
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 4de175ded7d..2b67f27e8b3 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
@@ -2131,8 +2131,6 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
Map<KeyCacheObject, CacheObject> putMap = null;
- Map<KeyCacheObject, EntryProcessor<Object, Object, Object>>
entryProcMap = null;
-
Collection<KeyCacheObject> rmvKeys = null;
List<CacheObject> writeVals = null;
@@ -2278,7 +2276,6 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
writeVals,
putMap,
null,
- entryProcMap,
req,
res,
replicate,
@@ -2291,7 +2288,6 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
putMap = null;
writeVals = null;
- entryProcMap = null;
filtered = new ArrayList<>();
}
@@ -2326,7 +2322,6 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
null,
null,
rmvKeys,
- entryProcMap,
req,
res,
replicate,
@@ -2338,7 +2333,6 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
firstEntryIdx = i;
rmvKeys = null;
- entryProcMap = null;
filtered = new ArrayList<>();
}
@@ -2351,11 +2345,6 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
putMap.put(entry.key(), updated);
writeVals.add(updated);
}
-
- if (entryProcMap == null)
- entryProcMap = new HashMap<>();
-
- entryProcMap.put(entry.key(), entryProc);
}
else if (op == UPDATE) {
CacheObject updated = req.value(i);
@@ -2450,7 +2439,6 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
writeVals,
putMap,
rmvKeys,
- entryProcMap,
req,
res,
replicate,
@@ -2620,13 +2608,10 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
else if (conflictCtx.isMerge())
newConflictVer = null; // Conflict version is
discarded in case of merge.
- EntryProcessor<Object, Object, Object> entryProc =
null;
-
dhtFut.addWriteEntry(
affAssignment,
entry,
updRes.newValue(),
- entryProc,
updRes.newTtl(),
updRes.conflictExpireTime(),
newConflictVer,
@@ -2642,7 +2627,6 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
readers,
entry,
updRes.newValue(),
- entryProc,
updRes.newTtl(),
updRes.conflictExpireTime(),
readRepairRecovery);
@@ -2744,7 +2728,6 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
* @param writeVals Write values.
* @param putMap Values to put.
* @param rmvKeys Keys to remove.
- * @param entryProcessorMap Entry processors.
* @param req Request.
* @param res Response.
* @param replicate Whether replication is enabled.
@@ -2762,7 +2745,6 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
@Nullable final List<CacheObject> writeVals,
@Nullable final Map<KeyCacheObject, CacheObject> putMap,
@Nullable final Collection<KeyCacheObject> rmvKeys,
- @Nullable final Map<KeyCacheObject, EntryProcessor<Object, Object,
Object>> entryProcessorMap,
final GridNearAtomicAbstractUpdateRequest req,
final GridNearAtomicUpdateResponse res,
final boolean replicate,
@@ -2860,9 +2842,6 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
// Get readers before innerUpdate (reader cleared after
remove).
GridDhtCacheEntry.ReaderId[] readers =
entry.readersLocked();
- EntryProcessor<Object, Object, Object> entryProc =
- entryProcessorMap == null ? null :
entryProcessorMap.get(entry.key());
-
boolean readRepairRecovery = op == TRANSFORM &&
req.entryProcessor(i) instanceof AtomicReadRepairEntryProcessor;
GridCacheUpdateAtomicResult updRes = entry.innerUpdate(
@@ -2895,7 +2874,7 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
null,
null,
dhtFut,
- entryProc != null);
+ req.operation() == TRANSFORM);
assert !updRes.success() || updRes.newTtl() ==
CU.TTL_NOT_CHANGED || expiry != null :
"success=" + updRes.success() + ", newTtl=" +
updRes.newTtl() + ", expiry=" + expiry;
@@ -2924,7 +2903,6 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
affAssignment,
entry,
writeVal,
- entryProc,
updRes.newTtl(),
CU.EXPIRE_TIME_CALCULATE,
null,
@@ -2940,7 +2918,6 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
readers,
entry,
writeVal,
- entryProc,
updRes.newTtl(),
CU.EXPIRE_TIME_CALCULATE,
readRepairRecovery);
@@ -3304,8 +3281,6 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
boolean replicate = ctx.isDrEnabled();
- boolean intercept = req.forceTransformBackups() &&
ctx.config().getInterceptor() != null;
-
boolean needTaskName =
ctx.events().isRecordable(EVT_CACHE_OBJECT_READ) ||
ctx.events().isRecordable(EVT_CACHE_OBJECT_PUT) ||
ctx.events().isRecordable(EVT_CACHE_OBJECT_REMOVED);
@@ -3328,11 +3303,9 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
CacheObject val = req.value(i);
CacheObject prevVal = req.previousValue(i);
- EntryProcessor<Object, Object, Object> entryProc =
req.entryProcessor(i);
Long updateIdx = req.updateCounter(i);
- GridCacheOperation op = entryProc != null ?
TRANSFORM :
- (val != null) ? UPDATE : DELETE;
+ GridCacheOperation op = val != null ? UPDATE :
DELETE;
long ttl = req.ttl(i);
long expireTime = req.conflictExpireTime(i);
@@ -3342,8 +3315,8 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
nodeId,
nodeId,
op,
- op == TRANSFORM ? entryProc : val,
- op == TRANSFORM ? req.invokeArguments() : null,
+ val,
+ null,
/*write-through*/(ctx.store().isLocal() &&
!ctx.shared().localStorePrimaryOnly())
&& writeThrough() && !req.skipStore(),
/*read-through*/false,
@@ -3354,7 +3327,7 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
/*event*/true,
/*metrics*/true,
/*primary*/false,
- /*check version*/!req.forceTransformBackups(),
+ /*check version*/true,
req.readRepairRecovery(),
req.topologyVersion(),
CU.empty0(),
@@ -3363,7 +3336,7 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
expireTime,
req.conflictVersion(i),
false,
- intercept,
+ /*intercept*/false,
taskName,
prevVal,
updateIdx,
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateFuture.java
index d93b5f790de..68a7d82a4d6 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateFuture.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateFuture.java
@@ -103,11 +103,9 @@ class GridDhtAtomicSingleUpdateFuture extends
GridDhtAtomicAbstractUpdateFuture
writeVer,
topVer,
updateReq.taskNameHash(),
- null,
updateReq.keepBinary(),
updateReq.keepBinaryInInterceptor(),
updateReq.skipStore(),
- false,
readRepairRecovery);
}
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateRequest.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateRequest.java
index d2787580889..c39c64ca8bb 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateRequest.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateRequest.java
@@ -18,7 +18,6 @@
package org.apache.ignite.internal.processors.cache.distributed.dht.atomic;
import java.util.UUID;
-import javax.cache.processor.EntryProcessor;
import org.apache.ignite.internal.Order;
import org.apache.ignite.internal.UseBinaryMarshaller;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
@@ -106,7 +105,6 @@ public class GridDhtAtomicSingleUpdateRequest extends
GridDhtAtomicAbstractUpdat
/**
* @param key Key to add.
* @param val Value, {@code null} if should be removed.
- * @param entryProc Entry processor.
* @param ttl TTL (optional).
* @param conflictExpireTime Conflict expire time (optional).
* @param conflictVer Conflict version (optional).
@@ -117,7 +115,6 @@ public class GridDhtAtomicSingleUpdateRequest extends
GridDhtAtomicAbstractUpdat
*/
@Override public void addWriteValue(KeyCacheObject key,
@Nullable CacheObject val,
- EntryProcessor<Object, Object, Object> entryProc,
long ttl,
long conflictExpireTime,
@Nullable GridCacheVersion conflictVer,
@@ -125,7 +122,6 @@ public class GridDhtAtomicSingleUpdateRequest extends
GridDhtAtomicAbstractUpdat
@Nullable CacheObject prevVal,
long updateCntr,
GridCacheOperation cacheOp) {
- assert entryProc == null;
assert ttl <= 0 : ttl;
assert conflictExpireTime <= 0 : conflictExpireTime;
assert conflictVer == null : conflictVer;
@@ -162,16 +158,13 @@ public class GridDhtAtomicSingleUpdateRequest extends
GridDhtAtomicAbstractUpdat
/**
* @param key Key to add.
* @param val Value, {@code null} if should be removed.
- * @param entryProc Entry processor.
* @param ttl TTL.
* @param expireTime Expire time.
*/
@Override public void addNearWriteValue(KeyCacheObject key,
@Nullable CacheObject val,
- EntryProcessor<Object, Object, Object> entryProc,
long ttl,
long expireTime) {
- assert entryProc == null;
assert ttl <= 0 : ttl;
assert key.partition() >= 0 : key;
@@ -187,11 +180,6 @@ public class GridDhtAtomicSingleUpdateRequest extends
GridDhtAtomicAbstractUpdat
this.val = val;
}
- /** {@inheritDoc} */
- @Override public boolean forceTransformBackups() {
- return false;
- }
-
/** {@inheritDoc} */
@Override public int size() {
return key != null ? near() ? 0 : 1 : 0;
@@ -265,18 +253,6 @@ public class GridDhtAtomicSingleUpdateRequest extends
GridDhtAtomicAbstractUpdat
return near() ? val : null;
}
- /** {@inheritDoc} */
- @Override @Nullable public EntryProcessor<Object, Object, Object>
entryProcessor(int idx) {
- assert idx == 0 : idx;
-
- return null;
- }
-
- /** {@inheritDoc} */
- @Override @Nullable public EntryProcessor<Object, Object, Object>
nearEntryProcessor(int idx) {
- return entryProcessor(idx);
- }
-
/** {@inheritDoc} */
@Override @Nullable public GridCacheVersion conflictVersion(int idx) {
assert idx == 0 : idx;
@@ -308,11 +284,6 @@ public class GridDhtAtomicSingleUpdateRequest extends
GridDhtAtomicAbstractUpdat
return conflictExpireTime(idx);
}
- /** {@inheritDoc} */
- @Override @Nullable public Object[] invokeArguments() {
- return null;
- }
-
/**
* Cleanup values not needed after message was sent.
*/
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
index 697c12dd24f..df4bb0584df 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
@@ -87,11 +87,9 @@ class GridDhtAtomicUpdateFuture extends
GridDhtAtomicAbstractUpdateFuture {
writeVer,
topVer,
updateReq.taskNameHash(),
- null,
updateReq.keepBinary(),
updateReq.keepBinaryInInterceptor(),
updateReq.skipStore(),
- false,
readRepairRecovery);
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
index 33b14d14d09..e36874727bf 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
@@ -20,17 +20,10 @@ package
org.apache.ignite.internal.processors.cache.distributed.dht.atomic;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
-import javax.cache.processor.EntryProcessor;
-import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.internal.MarshallableMessage;
import org.apache.ignite.internal.Order;
-import org.apache.ignite.internal.UseBinaryMarshaller;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheObject;
-import org.apache.ignite.internal.processors.cache.DeployableMessage;
-import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheOperation;
-import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.GridLongList;
@@ -38,15 +31,13 @@ import
org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.S;
-import org.apache.ignite.marshaller.Marshaller;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
/**
* Lite dht cache backup update request.
*/
-@UseBinaryMarshaller
-public class GridDhtAtomicUpdateRequest extends
GridDhtAtomicAbstractUpdateRequest implements MarshallableMessage,
DeployableMessage {
+public class GridDhtAtomicUpdateRequest extends
GridDhtAtomicAbstractUpdateRequest {
/** Keys to update. */
@GridToStringInclude
@Order(0)
@@ -97,33 +88,8 @@ public class GridDhtAtomicUpdateRequest extends
GridDhtAtomicAbstractUpdateReque
@Order(10)
List<Integer> obsoleteIndexes;
- /** Force transform backups flag. */
- @Order(11)
- boolean forceTransformBackups;
-
- /** Entry processors. */
- private List<EntryProcessor<Object, Object, Object>> entryProcessors;
-
- /** Entry processors bytes. */
- @Order(12)
- List<byte[]> entryProcessorsBytes;
-
- /** Near entry processors. */
- private List<EntryProcessor<Object, Object, Object>> nearEntryProcessors;
-
- /** Near entry processors bytes. */
- @Order(13)
- List<byte[]> nearEntryProcessorsBytes;
-
- /** Optional arguments for entry processor. */
- private Object[] invokeArgs;
-
- /** Entry processor arguments bytes. */
- @Order(14)
- List<byte[]> invokeArgsBytes;
-
/** Partition. */
- @Order(15)
+ @Order(11)
GridLongList updateCntrs;
/**
@@ -140,12 +106,10 @@ public class GridDhtAtomicUpdateRequest extends
GridDhtAtomicAbstractUpdateReque
* @param nodeId Node ID.
* @param futId Future ID.
* @param writeVer Write version for cache values.
- * @param invokeArgs Optional arguments for entry processor.
* @param topVer Topology version.
* @param keepBinary Keep binary flag.
* @param skipStore Skip store flag.
* @param keepBinaryInInterceptor Handle binary in interceptor operation
flag.
- * @param forceTransformBackups Force transform backups flag.
* @param taskNameHash Task name hash code.
* @param readRepairRecovery Recovery on Read Repair flag.
*/
@@ -156,11 +120,9 @@ public class GridDhtAtomicUpdateRequest extends
GridDhtAtomicAbstractUpdateReque
GridCacheVersion writeVer,
@NotNull AffinityTopologyVersion topVer,
int taskNameHash,
- Object[] invokeArgs,
boolean keepBinary,
boolean keepBinaryInInterceptor,
boolean skipStore,
- boolean forceTransformBackups,
boolean readRepairRecovery
) {
super(cacheId,
@@ -174,25 +136,13 @@ public class GridDhtAtomicUpdateRequest extends
GridDhtAtomicAbstractUpdateReque
skipStore,
readRepairRecovery);
- assert invokeArgs == null || forceTransformBackups;
-
- this.forceTransformBackups = forceTransformBackups;
- this.invokeArgs = invokeArgs;
-
keys = new ArrayList<>();
-
- if (forceTransformBackups) {
- entryProcessors = new ArrayList<>();
- entryProcessorsBytes = new ArrayList<>();
- }
- else
- vals = new ArrayList<>();
+ vals = new ArrayList<>();
}
/** {@inheritDoc} */
@Override public void addWriteValue(KeyCacheObject key,
@Nullable CacheObject val,
- EntryProcessor<Object, Object, Object> entryProc,
long ttl,
long conflictExpireTime,
@Nullable GridCacheVersion conflictVer,
@@ -203,14 +153,7 @@ public class GridDhtAtomicUpdateRequest extends
GridDhtAtomicAbstractUpdateReque
assert key.partition() >= 0 : key;
keys.add(key);
-
- if (forceTransformBackups) {
- assert entryProc != null;
-
- entryProcessors.add(entryProc);
- }
- else
- vals.add(val);
+ vals.add(val);
if (addPrevVal) {
if (prevVals == null)
@@ -262,7 +205,6 @@ public class GridDhtAtomicUpdateRequest extends
GridDhtAtomicAbstractUpdateReque
/** {@inheritDoc} */
@Override public void addNearWriteValue(KeyCacheObject key,
@Nullable CacheObject val,
- EntryProcessor<Object, Object, Object> entryProc,
long ttl,
long expireTime) {
assert key.partition() >= 0 : key;
@@ -280,24 +222,11 @@ public class GridDhtAtomicUpdateRequest extends
GridDhtAtomicAbstractUpdateReque
if (nearKeys == null) {
nearKeys = new ArrayList<>();
-
- if (forceTransformBackups) {
- nearEntryProcessors = new ArrayList<>();
- nearEntryProcessorsBytes = new ArrayList<>();
- }
- else
- nearVals = new ArrayList<>();
+ nearVals = new ArrayList<>();
}
nearKeys.add(key);
-
- if (forceTransformBackups) {
- assert entryProc != null;
-
- nearEntryProcessors.add(entryProc);
- }
- else
- nearVals.add(val);
+ nearVals.add(val);
if (ttl >= 0 && nearTtls == null) {
nearTtls = new GridLongList(nearKeys.size());
@@ -320,11 +249,6 @@ public class GridDhtAtomicUpdateRequest extends
GridDhtAtomicAbstractUpdateReque
nearExpireTimes.add(expireTime);
}
- /** {@inheritDoc} */
- @Override public boolean forceTransformBackups() {
- return forceTransformBackups;
- }
-
/** {@inheritDoc} */
@Override public int size() {
return keys.size();
@@ -379,11 +303,6 @@ public class GridDhtAtomicUpdateRequest extends
GridDhtAtomicAbstractUpdateReque
return null;
}
- /** {@inheritDoc} */
- @Override @Nullable public EntryProcessor<Object, Object, Object>
entryProcessor(int idx) {
- return entryProcessors == null ? null : entryProcessors.get(idx);
- }
-
/** {@inheritDoc} */
@Override @Nullable public CacheObject nearValue(int idx) {
if (nearVals != null)
@@ -392,11 +311,6 @@ public class GridDhtAtomicUpdateRequest extends
GridDhtAtomicAbstractUpdateReque
return null;
}
- /** {@inheritDoc} */
- @Override @Nullable public EntryProcessor<Object, Object, Object>
nearEntryProcessor(int idx) {
- return nearEntryProcessors == null ? null :
nearEntryProcessors.get(idx);
- }
-
/** {@inheritDoc} */
@Override @Nullable public GridCacheVersion conflictVersion(int idx) {
if (conflictVers != null) {
@@ -463,63 +377,12 @@ public class GridDhtAtomicUpdateRequest extends
GridDhtAtomicAbstractUpdateReque
return CU.EXPIRE_TIME_CALCULATE;
}
- /** {@inheritDoc} */
- @Override @Nullable public Object[] invokeArguments() {
- return invokeArgs;
- }
-
/** {@inheritDoc} */
@Override protected void cleanup() {
nearVals = null;
prevVals = null;
}
- /** {@inheritDoc} */
- @Override public void marshal(Marshaller marsh) throws
IgniteCheckedException {
- if (forceTransformBackups) {
- if (!F.isEmpty(invokeArgs) && invokeArgsBytes == null)
- invokeArgsBytes = marshallInvokeArguments(invokeArgs, marsh);
-
- if (entryProcessorsBytes == null)
- entryProcessorsBytes = marshallCollection(entryProcessors,
marsh);
-
- if (nearEntryProcessorsBytes == null)
- nearEntryProcessorsBytes =
marshallCollection(nearEntryProcessors, marsh);
- }
- }
-
- /** {@inheritDoc} */
- @Override public void unmarshal(Marshaller marsh, ClassLoader clsLdr)
throws IgniteCheckedException {
- if (forceTransformBackups) {
- if (entryProcessors == null)
- entryProcessors = unmarshalCollection(entryProcessorsBytes,
marsh, clsLdr);
-
- if (invokeArgsBytes != null && invokeArgs == null)
- invokeArgs = unmarshalInvokeArguments(invokeArgsBytes, marsh,
clsLdr);
-
- if (nearEntryProcessors == null)
- nearEntryProcessors =
unmarshalCollection(nearEntryProcessorsBytes, marsh, clsLdr);
- }
- }
-
- /** {@inheritDoc} */
- @Override public void deploy(GridCacheSharedContext<?, ?> ctx) throws
IgniteCheckedException {
- if (forceTransformBackups) {
- GridCacheContext<?, ?> cctx = ctx.cacheContext(cacheId);
-
- forceDeploymentInfo(ctx);
-
- if (!F.isEmpty(invokeArgs) && invokeArgsBytes == null)
- deployInvokeArguments(invokeArgs, cctx);
-
- if (entryProcessorsBytes == null)
- deployCollection(entryProcessors, cctx);
-
- if (nearEntryProcessorsBytes == null)
- deployCollection(nearEntryProcessors, cctx);
- }
- }
-
/** {@inheritDoc} */
@Override public String toString() {
return S.toString(GridDhtAtomicUpdateRequest.class, this, "super",
super.toString());
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
index 366f937e380..ee9509af199 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
@@ -319,8 +319,6 @@ public class GridNearAtomicCache<K, V> extends
GridNearCacheAdapter<K, V> {
assert ver != null;
- boolean intercept = req.forceTransformBackups() &&
ctx.config().getInterceptor() != null;
-
boolean needTaskName =
ctx.events().isRecordable(EVT_CACHE_OBJECT_READ) ||
ctx.events().isRecordable(EVT_CACHE_OBJECT_PUT) ||
ctx.events().isRecordable(EVT_CACHE_OBJECT_REMOVED);
@@ -347,10 +345,8 @@ public class GridNearAtomicCache<K, V> extends
GridNearCacheAdapter<K, V> {
}
CacheObject val = req.nearValue(i);
- EntryProcessor<Object, Object, Object> entryProc =
req.nearEntryProcessor(i);
- GridCacheOperation op = entryProc != null ? TRANSFORM :
- (val != null) ? UPDATE : DELETE;
+ GridCacheOperation op = val != null ? UPDATE : DELETE;
long ttl = req.nearTtl(i);
long expireTime = req.nearExpireTime(i);
@@ -360,8 +356,8 @@ public class GridNearAtomicCache<K, V> extends
GridNearCacheAdapter<K, V> {
nodeId,
nodeId,
op,
- op == TRANSFORM ? entryProc : val,
- op == TRANSFORM ? req.invokeArguments() : null,
+ val,
+ null,
/*write-through*/false,
/*read-through*/false,
/*retval*/false,
@@ -371,7 +367,7 @@ public class GridNearAtomicCache<K, V> extends
GridNearCacheAdapter<K, V> {
/*event*/true,
/*metrics*/true,
/*primary*/false,
- /*check version*/!req.forceTransformBackups(),
+ /*check version*/true,
false,
req.topologyVersion(),
CU.empty0(),
@@ -380,7 +376,7 @@ public class GridNearAtomicCache<K, V> extends
GridNearCacheAdapter<K, V> {
expireTime,
null,
false,
- intercept,
+ /*intercept*/false,
taskName,
null,
null,
