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 4927eec73bf IGNITE-28356 Get rid of redundant wire companion fields
and hand-written marshalling in messages (#13408)
4927eec73bf is described below
commit 4927eec73bff3cd81306feb5c070ee3f5cfcd3c1
Author: Anton Vinogradov <[email protected]>
AuthorDate: Thu Jul 30 21:19:36 2026 +0300
IGNITE-28356 Get rid of redundant wire companion fields and hand-written
marshalling in messages (#13408)
---
.../query/calcite/message/GenericValueMessage.java | 24 ++----
.../query/calcite/message/QueryStartRequest.java | 23 +-----
.../ignite/internal/CoreMessagesProvider.java | 2 +
.../ignite/internal/GridJobExecuteRequest.java | 64 ++++------------
.../GridDistributedTxPrepareRequest.java | 44 ++++-------
.../cache/distributed/dht/GridDhtTxLocal.java | 2 +-
.../distributed/dht/GridDhtTxPrepareFuture.java | 33 ++++-----
.../atomic/GridNearAtomicFullUpdateRequest.java | 27 +++----
.../cache/distributed/near/GridNearGetRequest.java | 46 +++---------
...dNearOptimisticSerializableTxPrepareFuture.java | 2 +-
.../near/GridNearOptimisticTxPrepareFuture.java | 2 +-
.../near/GridNearPessimisticTxPrepareFuture.java | 2 +-
.../cache/distributed/near/GridNearTxLocal.java | 2 +-
.../near/GridNearTxPrepareFutureAdapter.java | 8 +-
.../near/GridNearTxPrepareResponse.java | 47 ++++--------
.../cache/distributed/near/GridNearTxRemote.java | 8 +-
.../distributed/near/KeyedVersionedValue.java | 56 ++++++++++++++
.../continuous/CacheContinuousQueryEntry.java | 58 ++-------------
.../CacheContinuousQueryEventBuffer.java | 6 +-
.../continuous/CacheContinuousQueryHandler.java | 8 +-
.../cache/transactions/TxDeadlockDetection.java | 5 +-
.../cache/transactions/TxLocksRequest.java | 19 ++---
.../cache/transactions/TxLocksResponse.java | 19 +++--
.../internal/processors/job/GridJobProcessor.java | 9 ++-
.../query/schema/message/QueryEntityMessage.java | 28 ++-----
.../service/LazyServiceConfigurationMessage.java | 27 ++-----
.../internal/processors/task/GridTaskWorker.java | 2 -
.../near/GridNearTxPrepareDestroyedCacheTest.java | 20 ++---
...niteCacheContinuousQueryImmutableEntryTest.java | 85 ++++++++++++++--------
29 files changed, 274 insertions(+), 404 deletions(-)
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/GenericValueMessage.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/GenericValueMessage.java
index 774a89106e9..285c926f662 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/GenericValueMessage.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/GenericValueMessage.java
@@ -17,18 +17,17 @@
package org.apache.ignite.internal.processors.query.calcite.message;
-import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.internal.MarshallableMessage;
+import org.apache.ignite.internal.Marshalled;
import org.apache.ignite.internal.Order;
import org.apache.ignite.internal.UseBinaryMarshaller;
-import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.plugin.extensions.communication.Message;
/** */
@UseBinaryMarshaller
-public final class GenericValueMessage implements MarshallableMessage {
+public final class GenericValueMessage implements Message {
/** */
- private Object val;
+ @Marshalled("serialized")
+ Object val;
/** */
@Order(0)
@@ -49,17 +48,4 @@ public final class GenericValueMessage implements
MarshallableMessage {
return val;
}
- /** {@inheritDoc} */
- @Override public void marshal(Marshaller marsh) throws
IgniteCheckedException {
- if (val != null && serialized == null)
- serialized = U.marshal(marsh, val);
- }
-
- /** {@inheritDoc} */
- @Override public void unmarshal(Marshaller marsh, ClassLoader clsLdr)
throws IgniteCheckedException {
- if (serialized != null && val == null)
- val = U.unmarshal(marsh, serialized, clsLdr);
-
- serialized = null;
- }
}
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryStartRequest.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryStartRequest.java
index 8f35d8877fe..0ec34128837 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryStartRequest.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryStartRequest.java
@@ -20,20 +20,17 @@ package
org.apache.ignite.internal.processors.query.calcite.message;
import java.util.Collection;
import java.util.Map;
import java.util.UUID;
-import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.internal.DeferredUnmarshalMessage;
-import org.apache.ignite.internal.MarshallableMessage;
+import org.apache.ignite.internal.Marshalled;
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.query.calcite.metadata.FragmentDescription;
-import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.marshaller.Marshaller;
import org.jetbrains.annotations.Nullable;
/** Message sent to remote nodes to start a query fragment execution. */
@UseBinaryMarshaller
-public class QueryStartRequest implements MarshallableMessage,
DeferredUnmarshalMessage, ExecutionContextAware {
+public class QueryStartRequest implements DeferredUnmarshalMessage,
ExecutionContextAware {
/** */
@Order(0)
String schema;
@@ -63,7 +60,8 @@ public class QueryStartRequest implements
MarshallableMessage, DeferredUnmarshal
int totalFragmentsCnt;
/** */
- private @Nullable Object[] params;
+ @Marshalled("paramsBytes")
+ @Nullable Object[] params;
/** */
@Order(7)
@@ -214,17 +212,4 @@ public class QueryStartRequest implements
MarshallableMessage, DeferredUnmarshal
return keepBinaryMode;
}
- /** {@inheritDoc} */
- @Override public void marshal(Marshaller marsh) throws
IgniteCheckedException {
- if (paramsBytes == null && params != null)
- paramsBytes = U.marshal(marsh, params);
- }
-
- /** {@inheritDoc} */
- @Override public void unmarshal(Marshaller marsh, ClassLoader clsLdr)
throws IgniteCheckedException {
- if (params == null && paramsBytes != null)
- params = U.unmarshal(marsh, paramsBytes, clsLdr);
-
- paramsBytes = null;
- }
}
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 ad61d016b8c..7921779c09e 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
@@ -150,6 +150,7 @@ import
org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxFi
import
org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxFinishResponse;
import
org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareRequest;
import
org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareResponse;
+import
org.apache.ignite.internal.processors.cache.distributed.near.KeyedVersionedValue;
import
org.apache.ignite.internal.processors.cache.persistence.snapshot.DataStreamerUpdatesHandlerResult;
import
org.apache.ignite.internal.processors.cache.persistence.snapshot.IncrementalSnapshotAwareMessage;
import
org.apache.ignite.internal.processors.cache.persistence.snapshot.IncrementalSnapshotVerifyResult;
@@ -372,6 +373,7 @@ public class CoreMessagesProvider extends
AbstractMarshallableMessageFactoryProv
withNoSchema(DiscoveryDataPacket.class);
withNoSchema(GridByteArrayList.class);
withNoSchema(CacheVersionedValue.class);
+ withNoSchema(KeyedVersionedValue.class);
withNoSchema(WALPointer.class);
withNoSchema(SerializableDataBagItemWrapper.class);
withSchema(GridTopicMessage.class);
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/GridJobExecuteRequest.java
b/modules/core/src/main/java/org/apache/ignite/internal/GridJobExecuteRequest.java
index f08584b10d7..098e8514f68 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/GridJobExecuteRequest.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/GridJobExecuteRequest.java
@@ -21,7 +21,6 @@ import java.io.Serializable;
import java.util.Collection;
import java.util.Map;
import java.util.UUID;
-import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.compute.ComputeJob;
import org.apache.ignite.compute.ComputeJobSibling;
@@ -33,7 +32,6 @@ import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.lang.IgniteUuid;
-import org.apache.ignite.marshaller.Marshaller;
import org.jetbrains.annotations.Nullable;
/**
@@ -41,7 +39,7 @@ import org.jetbrains.annotations.Nullable;
*/
@SuppressWarnings({"AssignmentOrReturnOfFieldWithMutableType",
"NullableProblems"})
@UseBinaryMarshaller
-public class GridJobExecuteRequest implements ExecutorAwareMessage {
+public class GridJobExecuteRequest implements ExecutorAwareMessage,
DeferredUnmarshalMessage {
/** */
@Order(0)
IgniteUuid sesId;
@@ -57,7 +55,8 @@ public class GridJobExecuteRequest implements
ExecutorAwareMessage {
/** */
@GridToStringExclude
- private ComputeJob job;
+ @Marshalled("jobBytes")
+ ComputeJob job;
/** */
@Order(3)
@@ -91,7 +90,8 @@ public class GridJobExecuteRequest implements
ExecutorAwareMessage {
/** */
@GridToStringExclude
- private Map<Object, Object> sesAttrs;
+ @Marshalled("sesAttrsBytes")
+ Map<Object, Object> sesAttrs;
/** */
@GridToStringExclude
@@ -100,14 +100,16 @@ public class GridJobExecuteRequest implements
ExecutorAwareMessage {
/** */
@GridToStringExclude
- private Map<? extends Serializable, ? extends Serializable> jobAttrs;
+ @Marshalled("jobAttrsBytes")
+ Map<? extends Serializable, ? extends Serializable> jobAttrs;
/** Checkpoint SPI name. */
@Order(11)
String cpSpi;
- /** */
- private Collection<ComputeJobSibling> siblings;
+ /** Left unset for a continuous task: such a job requests its siblings
from the task node instead. */
+ @Marshalled("siblingsBytes")
+ Collection<ComputeJobSibling> siblings;
/** */
@Order(12)
@@ -145,7 +147,8 @@ public class GridJobExecuteRequest implements
ExecutorAwareMessage {
Collection<UUID> top;
/** */
- private IgnitePredicate<ClusterNode> topPred;
+ @Marshalled("topPredBytes")
+ IgnitePredicate<ClusterNode> topPred;
/** */
@Order(20)
@@ -250,7 +253,7 @@ public class GridJobExecuteRequest implements
ExecutorAwareMessage {
this.top = top;
this.topVer = topVer;
this.topPred = topPred;
- this.siblings = siblings;
+ this.siblings = dynamicSiblings ? null : siblings;
this.sesAttrs = sesAttrs;
this.jobAttrs = jobAttrs;
this.clsLdrId = clsLdrId;
@@ -449,45 +452,4 @@ public class GridJobExecuteRequest implements
ExecutorAwareMessage {
return S.toString(GridJobExecuteRequest.class, this);
}
- /**
- * @param marsh Marshaller.
- */
- public void marshal(Marshaller marsh) throws IgniteCheckedException {
- jobBytes = U.marshal(marsh, job);
- topPredBytes = U.marshal(marsh, topPred);
- siblingsBytes = U.marshal(marsh, siblings);
- sesAttrsBytes = U.marshal(marsh, sesAttrs);
- jobAttrsBytes = U.marshal(marsh, jobAttrs);
- }
-
- /**
- * @param marsh Marshaller.
- * @param ldr Class loader.
- */
- public void unmarshal(Marshaller marsh, ClassLoader ldr) throws
IgniteCheckedException {
- assert top != null || topPredBytes != null;
- assert sesAttrsBytes != null || !sesFullSup;
-
- if (!dynamicSiblings && siblings == null)
- siblings = U.unmarshal(marsh, siblingsBytes, ldr);
-
- if (sesFullSup && sesAttrs == null)
- sesAttrs = U.unmarshal(marsh, sesAttrsBytes, ldr);
-
- if (topPred == null && topPredBytes != null)
- topPred = U.unmarshal(marsh, topPredBytes, ldr);
-
- if (jobAttrs == null)
- jobAttrs = U.unmarshal(marsh, jobAttrsBytes, ldr);
-
- if (job == null)
- job = U.unmarshal(marsh, jobBytes, ldr);
-
- // Are not required anymore.
- siblingsBytes = null;
- sesAttrsBytes = null;
- topPredBytes = null;
- jobAttrsBytes = null;
- jobBytes = null;
- }
}
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 9d04dd9ff26..7820e58513e 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
@@ -17,13 +17,11 @@
package org.apache.ignite.internal.processors.cache.distributed;
+import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import org.apache.ignite.IgniteLogger;
-import org.apache.ignite.internal.Marshalled;
import org.apache.ignite.internal.Order;
import org.apache.ignite.internal.UseBinaryMarshaller;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
@@ -36,6 +34,7 @@ import
org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.tostring.GridToStringBuilder;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.transactions.TransactionConcurrency;
import org.apache.ignite.transactions.TransactionIsolation;
import org.jetbrains.annotations.Nullable;
@@ -99,36 +98,28 @@ public class GridDistributedTxPrepareRequest extends
GridDistributedBaseMessage
@GridToStringInclude
public Collection<IgniteTxEntry> writes;
- /** DHT versions to verify. */
- @GridToStringInclude
- @Marshalled(keys = "dhtVerKeys", values = "dhtVerVals")
- public Map<IgniteTxKey, GridCacheVersion> dhtVers;
-
- /** */
+ /** Keys whose DHT version has to be verified on the remote node. */
@Order(7)
+ @GridToStringInclude
public Collection<IgniteTxKey> dhtVerKeys;
- /** */
- @Order(8)
- public Collection<GridCacheVersion> dhtVerVals;
-
/** Expected transaction size. */
- @Order(9)
+ @Order(8)
public int txSize;
/** Transaction nodes mapping (primary node -> related backup nodes). */
- @Order(10)
+ @Order(9)
public Map<UUID, Collection<UUID>> txNodes;
/** IO policy. */
- @Order(11)
+ @Order(10)
public byte plc;
/** Transient TX state. */
private IgniteTxState txState;
/** */
- @Order(12)
+ @Order(11)
@GridToStringExclude
public byte flags;
@@ -237,23 +228,20 @@ public class GridDistributedTxPrepareRequest extends
GridDistributedBaseMessage
}
/**
- * Adds version to be verified on remote node.
- *
- * @param key Key for which version is verified.
- * @param dhtVer DHT version to check.
+ * @param key Key whose DHT version is verified on the remote node.
*/
- public void addDhtVersion(IgniteTxKey key, @Nullable GridCacheVersion
dhtVer) {
- if (dhtVers == null)
- dhtVers = new HashMap<>();
+ public void addDhtVersionKey(IgniteTxKey key) {
+ if (dhtVerKeys == null)
+ dhtVerKeys = new ArrayList<>();
- dhtVers.put(key, dhtVer);
+ dhtVerKeys.add(key);
}
/**
- * @return Map of versions to be verified.
+ * @return Keys whose DHT version is verified.
*/
- public Map<IgniteTxKey, GridCacheVersion> dhtVersions() {
- return dhtVers == null ? Collections.emptyMap() : dhtVers;
+ public Collection<IgniteTxKey> dhtVersionKeys() {
+ return F.emptyIfNull(dhtVerKeys);
}
/**
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java
index 67b32d38f86..d05e7592955 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java
@@ -325,7 +325,7 @@ public class GridDhtTxLocal extends GridDhtTxLocalAdapter
implements GridCacheMa
this,
timeout,
req.miniId(),
- req.dhtVersions(),
+ req.dhtVersionKeys(),
req.last(),
needReturnValue()))) {
GridDhtTxPrepareFuture f = prepFut;
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
index 5ee08cc4774..f2fc3422292 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
@@ -183,8 +183,8 @@ public final class GridDhtTxPrepareFuture extends
GridCacheCompoundFuture<Ignite
/** Near mini future id. */
private int nearMiniId;
- /** DHT versions map. */
- private Map<IgniteTxKey, GridCacheVersion> dhtVerMap;
+ /** DHT version keys. */
+ private Collection<IgniteTxKey> dhtVerKeys;
/** {@code True} if this is last prepare operation for node. */
private boolean last;
@@ -226,7 +226,7 @@ public final class GridDhtTxPrepareFuture extends
GridCacheCompoundFuture<Ignite
* @param tx Transaction.
* @param timeout Timeout.
* @param nearMiniId Near mini future id.
- * @param dhtVerMap DHT versions map.
+ * @param dhtVerKeys Keys whose DHT version is verified.
* @param last {@code True} if this is last prepare operation for node.
* @param retVal Return value flag.
*/
@@ -235,7 +235,7 @@ public final class GridDhtTxPrepareFuture extends
GridCacheCompoundFuture<Ignite
final GridDhtTxLocalAdapter tx,
long timeout,
int nearMiniId,
- Map<IgniteTxKey, GridCacheVersion> dhtVerMap,
+ Collection<IgniteTxKey> dhtVerKeys,
boolean last,
boolean retVal
) {
@@ -243,7 +243,7 @@ public final class GridDhtTxPrepareFuture extends
GridCacheCompoundFuture<Ignite
this.cctx = cctx;
this.tx = tx;
- this.dhtVerMap = dhtVerMap;
+ this.dhtVerKeys = dhtVerKeys;
this.last = last;
this.deploymentLdrId =
U.contextDeploymentClassLoaderId(cctx.kernalContext());
@@ -941,6 +941,8 @@ public final class GridDhtTxPrepareFuture extends
GridCacheCompoundFuture<Ignite
* @param res Response being sent.
*/
private void addDhtValues(GridNearTxPrepareResponse res) {
+ Collection<IgniteTxKey> added = new HashSet<>();
+
// Interceptor on near node needs old values to execute callbacks.
if (req.writes() != null) {
for (IgniteTxEntry e : req.writes()) {
@@ -958,9 +960,12 @@ public final class GridDhtTxPrepareFuture extends
GridCacheCompoundFuture<Ignite
CacheObject val0 = entry.valueBytes();
- if (val0 != null)
+ if (val0 != null) {
res.addOwnedValue(txEntry.txKey(), dhtVer, val0);
+ added.add(txEntry.txKey());
+ }
+
break;
}
catch (GridCacheEntryRemovedException ignored) {
@@ -971,13 +976,13 @@ public final class GridDhtTxPrepareFuture extends
GridCacheCompoundFuture<Ignite
}
}
- for (Map.Entry<IgniteTxKey, GridCacheVersion> ver :
dhtVerMap.entrySet()) {
- IgniteTxEntry txEntry = tx.entry(ver.getKey());
+ for (IgniteTxKey key : dhtVerKeys) {
+ IgniteTxEntry txEntry = tx.entry(key);
- if (res.hasOwnedValue(ver.getKey()))
+ if (added.contains(key))
continue;
- assert txEntry != null : ver;
+ assert txEntry != null : key;
GridCacheContext cacheCtx = txEntry.context();
@@ -985,13 +990,7 @@ public final class GridDhtTxPrepareFuture extends
GridCacheCompoundFuture<Ignite
try {
GridCacheEntryEx entry = txEntry.cached();
- GridCacheVersion dhtVer = entry.version();
-
- if (ver.getValue() == null ||
!ver.getValue().equals(dhtVer)) {
- CacheObject val0 = entry.valueBytes();
-
- res.addOwnedValue(txEntry.txKey(), dhtVer, val0);
- }
+ res.addOwnedValue(txEntry.txKey(), entry.version(),
entry.valueBytes());
break;
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicFullUpdateRequest.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicFullUpdateRequest.java
index ac095f09b0d..cc868b91bb6 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicFullUpdateRequest.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicFullUpdateRequest.java
@@ -26,6 +26,7 @@ import javax.cache.processor.EntryProcessor;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cache.CacheWriteSynchronizationMode;
import org.apache.ignite.internal.MarshallableMessage;
+import org.apache.ignite.internal.Marshalled;
import org.apache.ignite.internal.Order;
import org.apache.ignite.internal.UseBinaryMarshaller;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
@@ -67,7 +68,8 @@ public class GridNearAtomicFullUpdateRequest extends
GridNearAtomicAbstractUpdat
List<CacheObject> vals;
/** Entry processors. */
- private List<EntryProcessor<Object, Object, Object>> entryProcessors;
+ @Marshalled("entryProcessorsBytes")
+ List<Object> entryProcessors;
/** Entry processors bytes. */
@Order(2)
@@ -268,7 +270,7 @@ public class GridNearAtomicFullUpdateRequest extends
GridNearAtomicAbstractUpdat
@Override public EntryProcessor<Object, Object, Object> entryProcessor(int
idx) {
assert operation() == TRANSFORM : operation();
- return entryProcessors.get(idx);
+ return (EntryProcessor<Object, Object,
Object>)entryProcessors.get(idx);
}
/** {@inheritDoc} */
@@ -355,14 +357,9 @@ public class GridNearAtomicFullUpdateRequest extends
GridNearAtomicAbstractUpdat
@Override public void marshal(Marshaller marsh) throws
IgniteCheckedException {
if (expiryPlc != null && expiryPlcBytes == null)
expiryPlcBytes = U.marshal(marsh, new
IgniteExternalizableExpiryPolicy(expiryPlc));
-
- if (operation() == TRANSFORM) {
- if (entryProcessorsBytes == null)
- entryProcessorsBytes = marshallCollection(entryProcessors,
marsh);
- if (!F.isEmpty(invokeArgs) && invokeArgsBytes == null)
- invokeArgsBytes = marshallInvokeArguments(invokeArgs, marsh);
- }
+ if (operation() == TRANSFORM && !F.isEmpty(invokeArgs) &&
invokeArgsBytes == null)
+ invokeArgsBytes = marshallInvokeArguments(invokeArgs, marsh);
}
/** {@inheritDoc} */
@@ -370,13 +367,8 @@ public class GridNearAtomicFullUpdateRequest extends
GridNearAtomicAbstractUpdat
if (expiryPlcBytes != null && expiryPlc == null)
expiryPlc = U.unmarshal(marsh, expiryPlcBytes, clsLdr);
- if (operation() == TRANSFORM) {
- if (entryProcessors == null)
- entryProcessors = unmarshalCollection(entryProcessorsBytes,
marsh, clsLdr);
-
- if (invokeArgsBytes != null && invokeArgs == null)
- invokeArgs = unmarshalInvokeArguments(invokeArgsBytes, marsh,
clsLdr);
- }
+ if (operation() == TRANSFORM && invokeArgsBytes != null && invokeArgs
== null)
+ invokeArgs = unmarshalInvokeArguments(invokeArgsBytes, marsh,
clsLdr);
}
/** {@inheritDoc} */
@@ -389,7 +381,8 @@ public class GridNearAtomicFullUpdateRequest extends
GridNearAtomicAbstractUpdat
forceDeploymentInfo(ctx);
- if (entryProcessorsBytes == null)
+ // Gated on the object, not on the bytes: the generated unmarshal
nulls the companion once it is done.
+ if (entryProcessors != null)
deployCollection(entryProcessors, cctx);
if (!F.isEmpty(invokeArgs) && invokeArgsBytes == null)
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java
index ef0e33772fc..7e4c199dacb 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java
@@ -17,11 +17,9 @@
package org.apache.ignite.internal.processors.cache.distributed.near;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.LinkedHashMap;
-import java.util.List;
+import java.util.Collection;
import java.util.Map;
+import org.apache.ignite.internal.Marshalled;
import org.apache.ignite.internal.Order;
import org.apache.ignite.internal.UseBinaryMarshaller;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
@@ -33,7 +31,6 @@ import
org.apache.ignite.internal.processors.cache.version.GridCacheVersionable;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.S;
-import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteUuid;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -69,15 +66,16 @@ public class GridNearGetRequest extends GridCacheIdMessage
implements GridCacheD
/** */
@GridToStringInclude
- private LinkedHashMap<KeyCacheObject, Boolean> keyMap;
+ @Marshalled(keys = "keys", values = "readersFlags")
+ Map<KeyCacheObject, Boolean> keyMap;
/** */
@Order(3)
- List<KeyCacheObject> keys;
+ Collection<KeyCacheObject> keys;
/** */
@Order(4)
- List<Boolean> readersFlags;
+ Collection<Boolean> readersFlags;
/** */
@Order(5)
@@ -149,19 +147,7 @@ public class GridNearGetRequest extends GridCacheIdMessage
implements GridCacheD
this.futId = futId;
this.miniId = miniId;
this.ver = ver;
-
- this.keys = new ArrayList<>(keys.size());
-
- if (addReader)
- readersFlags = new ArrayList<>(keys.size());
-
- for (Map.Entry<KeyCacheObject, Boolean> entry : keys.entrySet()) {
- this.keys.add(entry.getKey());
-
- if (addReader)
- readersFlags.add(entry.getValue());
- }
-
+ keyMap = keys;
this.topVer = topVer;
this.taskNameHash = taskNameHash;
this.createTtl = createTtl;
@@ -212,19 +198,7 @@ public class GridNearGetRequest extends GridCacheIdMessage
implements GridCacheD
/**
* @return Keys.
*/
- public LinkedHashMap<KeyCacheObject, Boolean> keyMap() {
- if (keyMap == null && !F.isEmpty(keys)) {
- keyMap = U.newLinkedHashMap(keys.size());
-
- Iterator<KeyCacheObject> keysIt = keys.iterator();
-
- for (int i = 0; i < keys.size(); i++) {
- Boolean addRdr = readersFlags != null ? readersFlags.get(i) :
Boolean.FALSE;
-
- keyMap.put(keysIt.next(), addRdr);
- }
- }
-
+ public Map<KeyCacheObject, Boolean> keyMap() {
return keyMap;
}
@@ -278,7 +252,9 @@ public class GridNearGetRequest extends GridCacheIdMessage
implements GridCacheD
/** {@inheritDoc} */
@Override public int partition() {
- return keys != null && !keys.isEmpty() ? keys.get(0).partition() : -1;
+ Collection<KeyCacheObject> keys0 = keyMap != null ? keyMap.keySet() :
keys;
+
+ return F.isEmpty(keys0) ? -1 : keys0.iterator().next().partition();
}
/**
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java
index a7fab50eaf3..89692e63737 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java
@@ -534,7 +534,7 @@ public class GridNearOptimisticSerializableTxPrepareFuture
extends GridNearOptim
for (IgniteTxEntry txEntry : writes) {
if (txEntry.op() == TRANSFORM)
- req.addDhtVersion(txEntry.txKey(), null);
+ req.addDhtVersionKey(txEntry.txKey());
}
req.miniId(fut.futureId());
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
index 2cf1f93e348..ad45b740ec6 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
@@ -528,7 +528,7 @@ public class GridNearOptimisticTxPrepareFuture extends
GridNearOptimisticTxPrepa
for (IgniteTxEntry txEntry : m.entries()) {
if (txEntry.op() == TRANSFORM)
- req.addDhtVersion(txEntry.txKey(), null);
+ req.addDhtVersionKey(txEntry.txKey());
}
// Must lock near entries separately.
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
index 059827ed4dd..1da972e57d5 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
@@ -231,7 +231,7 @@ public class GridNearPessimisticTxPrepareFuture extends
GridNearTxPrepareFutureA
for (IgniteTxEntry txEntry : writes) {
if (txEntry.op() == TRANSFORM)
- req.addDhtVersion(txEntry.txKey(), null);
+ req.addDhtVersionKey(txEntry.txKey());
}
return req;
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
index 1ecbd0713ae..ad3dbc70824 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
@@ -4079,7 +4079,7 @@ public class GridNearTxLocal extends
GridDhtTxLocalAdapter implements GridTimeou
this,
timeout,
0,
- Collections.emptyMap(),
+ Collections.emptyList(),
req.last(),
needReturnValue() && implicit());
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFutureAdapter.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFutureAdapter.java
index 4c6ace62bf1..db470fbd779 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFutureAdapter.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFutureAdapter.java
@@ -209,8 +209,8 @@ public abstract class GridNearTxPrepareFutureAdapter extends
UUID nodeId = m.primary().id();
- for (Map.Entry<IgniteTxKey, CacheVersionedValue> entry :
res.ownedValues().entrySet()) {
- IgniteTxEntry txEntry = tx.entry(entry.getKey());
+ for (KeyedVersionedValue tup : res.ownedValues()) {
+ IgniteTxEntry txEntry = tx.entry(tup.txKey());
assert txEntry != null;
@@ -221,8 +221,6 @@ public abstract class GridNearTxPrepareFutureAdapter extends
if (cacheCtx.isNear()) {
GridNearCacheEntry nearEntry =
(GridNearCacheEntry)txEntry.cached();
- CacheVersionedValue tup = entry.getValue();
-
nearEntry.resetFromPrimary(tup.value(),
tx.xidVersion(),
tup.version(),
@@ -232,8 +230,6 @@ public abstract class GridNearTxPrepareFutureAdapter extends
else if (txEntry.cached().detached()) {
GridDhtDetachedCacheEntry detachedEntry =
(GridDhtDetachedCacheEntry)txEntry.cached();
- CacheVersionedValue tup = entry.getValue();
-
detachedEntry.resetFromPrimary(tup.value(),
tx.xidVersion());
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareResponse.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareResponse.java
index 0150ee99cd5..bd26d752f67 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareResponse.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareResponse.java
@@ -17,11 +17,9 @@
package org.apache.ignite.internal.processors.cache.distributed.near;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import org.apache.ignite.internal.Marshalled;
import org.apache.ignite.internal.Order;
import org.apache.ignite.internal.UseBinaryMarshaller;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
@@ -31,7 +29,6 @@ import
org.apache.ignite.internal.processors.cache.distributed.GridDistributedTx
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
-import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.lang.IgniteUuid;
import org.jetbrains.annotations.Nullable;
@@ -62,33 +59,25 @@ public class GridNearTxPrepareResponse extends
GridDistributedTxPrepareResponse
@Order(4)
GridCacheVersion writeVer;
- /** Map of owned values to set on near node. */
- @GridToStringInclude
- @Marshalled(keys = "ownedValKeys", values = "ownedValVals")
- Map<IgniteTxKey, CacheVersionedValue> ownedVals;
-
- /** OwnedVals' keys for marshalling. */
+ /** Owned values to set on near node. */
@Order(5)
- @Nullable Collection<IgniteTxKey> ownedValKeys;
-
- /** OwnedVals' values for marshalling. */
- @Order(6)
- @Nullable Collection<CacheVersionedValue> ownedValVals;
+ @GridToStringInclude
+ @Nullable Collection<KeyedVersionedValue> ownedVals;
/** Cache return value. */
- @Order(7)
+ @Order(6)
GridCacheReturn retVal;
/** Keys that did not pass the filter. */
- @Order(8)
+ @Order(7)
@Nullable Collection<IgniteTxKey> filterFailedKeys;
/** Topology version, which is set when client node should remap lock
request. */
- @Order(9)
+ @Order(8)
@Nullable AffinityTopologyVersion clientRemapVer;
/** One-phase commit on primary flag. */
- @Order(10)
+ @Order(9)
boolean onePhaseCommit;
/**
@@ -193,18 +182,16 @@ public class GridNearTxPrepareResponse extends
GridDistributedTxPrepareResponse
return;
if (ownedVals == null)
- ownedVals = new HashMap<>();
+ ownedVals = new ArrayList<>();
- CacheVersionedValue oVal = new CacheVersionedValue(val, ver,
key.cacheId());
-
- ownedVals.put(key, oVal);
+ ownedVals.add(new KeyedVersionedValue(key, val, ver));
}
/**
- * @return Map of owned values to set on near node.
+ * @return Owned values to set on near node.
*/
- public Map<IgniteTxKey, CacheVersionedValue> ownedValues() {
- return ownedVals == null ? Collections.emptyMap() :
Collections.unmodifiableMap(ownedVals);
+ public Collection<KeyedVersionedValue> ownedValues() {
+ return ownedVals == null ? Collections.emptyList() :
Collections.unmodifiableCollection(ownedVals);
}
/** @return Cache return value. */
@@ -226,14 +213,6 @@ public class GridNearTxPrepareResponse extends
GridDistributedTxPrepareResponse
return filterFailedKeys;
}
- /**
- * @param key Key.
- * @return {@code True} if response has owned value for given key.
- */
- public boolean hasOwnedValue(IgniteTxKey key) {
- return F.mapContainsKey(ownedVals, key);
- }
-
/** {@inheritDoc} */
@Override public String toString() {
return S.toString(GridNearTxPrepareResponse.class, this, "super",
super.toString());
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxRemote.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxRemote.java
index 6023dbd144c..42ca9f8fa8e 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxRemote.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxRemote.java
@@ -36,7 +36,6 @@ import
org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
import
org.apache.ignite.internal.processors.cache.transactions.IgniteTxRemoteStateImpl;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
-import org.apache.ignite.internal.util.GridLeanMap;
import org.apache.ignite.internal.util.tostring.GridToStringBuilder;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -177,10 +176,11 @@ public class GridNearTxRemote extends
GridDistributedTxRemoteAdapter {
if (F.isEmpty(vers))
return;
+ // The request is done with by now, so its map is taken over instead
of being copied.
if (owned == null)
- owned = new GridLeanMap<>(vers.size());
-
- owned.putAll(vers);
+ owned = vers;
+ else
+ owned.putAll(vers);
}
/** {@inheritDoc} */
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/KeyedVersionedValue.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/KeyedVersionedValue.java
new file mode 100644
index 00000000000..793fb8cf59b
--- /dev/null
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/KeyedVersionedValue.java
@@ -0,0 +1,56 @@
+/*
+ * 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.near;
+
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.processors.cache.CacheObject;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.internal.S;
+
+/** Cache object and version, told apart by the key they belong to. */
+public class KeyedVersionedValue extends CacheVersionedValue {
+ /** */
+ @Order(0)
+ @GridToStringInclude
+ KeyCacheObject key;
+
+ /** */
+ public KeyedVersionedValue() {
+ // No-op.
+ }
+
+ /** */
+ public KeyedVersionedValue(IgniteTxKey txKey, CacheObject val,
GridCacheVersion ver) {
+ super(val, ver, txKey.cacheId());
+
+ key = txKey.key();
+ }
+
+ /** @return Key this value belongs to. */
+ public IgniteTxKey txKey() {
+ return new IgniteTxKey(key, cacheId());
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(KeyedVersionedValue.class, this, "super",
super.toString());
+ }
+}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntry.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntry.java
index 85bb7840402..841230e73f8 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntry.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntry.java
@@ -18,8 +18,6 @@
package org.apache.ignite.internal.processors.cache.query.continuous;
import javax.cache.event.EventType;
-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.managers.deployment.GridDeploymentInfo;
@@ -30,15 +28,15 @@ import
org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.internal.S;
-import org.apache.ignite.marshaller.Marshaller;
import org.apache.ignite.plugin.extensions.communication.CacheIdAware;
+import org.apache.ignite.plugin.extensions.communication.Message;
import org.jetbrains.annotations.Nullable;
/**
* Continuous query entry.
*/
@UseBinaryMarshaller
-public class CacheContinuousQueryEntry implements GridCacheDeployable,
MarshallableMessage, CacheIdAware {
+public class CacheContinuousQueryEntry implements GridCacheDeployable,
Message, CacheIdAware {
/** */
private static final byte BACKUP_ENTRY = 0b0001;
@@ -56,30 +54,21 @@ public class CacheContinuousQueryEntry implements
GridCacheDeployable, Marshalla
@Order(1)
byte flags;
- /** Key. */
+ /** Key. {@code null} for a filtered entry. */
+ @Order(2)
@GridToStringInclude
KeyCacheObject key;
- /** {@code null} for a filtered entry. */
- @Order(2)
- KeyCacheObject keyWire;
-
- /** New value. */
+ /** New value. {@code null} for a filtered entry. */
+ @Order(3)
@GridToStringInclude
CacheObject newVal;
- /** {@code null} for a filtered entry. */
- @Order(3)
- CacheObject newValWire;
-
- /** Old value. */
+ /** Old value. {@code null} for a filtered entry. */
+ @Order(4)
@GridToStringInclude
CacheObject oldVal;
- /** {@code null} for a filtered entry. */
- @Order(4)
- CacheObject oldValWire;
-
/** Cache name. */
@Order(5)
int cacheId;
@@ -326,37 +315,6 @@ public class CacheContinuousQueryEntry implements
GridCacheDeployable, Marshalla
return depInfo;
}
- // TODO IGNITE-28922: get rid of the wire-companion field pairs.
- /** {@inheritDoc} */
- @Override public void marshal(Marshaller marsh) throws
IgniteCheckedException {
- if (!isFiltered()) {
- keyWire = key;
- newValWire = newVal;
- oldValWire = oldVal;
- }
- }
-
- /** {@inheritDoc} */
- @Override public void unmarshal(Marshaller marsh, ClassLoader clsLdr)
throws IgniteCheckedException {
- if (keyWire != null) {
- key = keyWire;
-
- keyWire = null;
- }
-
- if (newValWire != null) {
- newVal = newValWire;
-
- newValWire = null;
- }
-
- if (oldValWire != null) {
- oldVal = oldValWire;
-
- oldValWire = null;
- }
- }
-
/** {@inheritDoc} */
@Override public String toString() {
return S.toString(CacheContinuousQueryEntry.class, this);
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
index 01b0ae67321..9d4b75a213e 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java
@@ -178,7 +178,7 @@ public class CacheContinuousQueryEventBuffer {
* @return Collected entries to pass to listener (single entry or entries
list).
*/
@Nullable Object processEntry(CacheContinuousQueryEntry e, boolean backup)
{
- return process0(e.updateCounter(), e, backup);
+ return process0(e.updateCounter(), e.copyWithDataReset(), backup);
}
/**
@@ -236,7 +236,7 @@ public class CacheContinuousQueryEventBuffer {
// Discard messages on backup and send to client
if primary.
if (!backup)
- res = addResult(res,
entry0.copyWithDataReset(), backup);
+ res = addResult(res, entry0, backup);
iter.remove();
pendingCurrSize.decrementAndGet();
@@ -490,8 +490,6 @@ public class CacheContinuousQueryEventBuffer {
if (entries == null)
return RETRY;
- entry = entry.copyWithDataReset();
-
entries[pos] = entry;
int next = lastProc + 1;
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
index 30668521d71..805ebda1e00 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
@@ -1332,7 +1332,7 @@ public class CacheContinuousQueryHandler<K, V> implements
GridContinuousHandler
CacheContinuousQueryEventBuffer buf = partitionBuffer(cctx,
e.partition());
- buf.processEntry(e.copyWithDataReset(), true);
+ buf.processEntry(e, true);
}
/**
@@ -1351,10 +1351,10 @@ public class CacheContinuousQueryHandler<K, V>
implements GridContinuousHandler
return e;
}
- // Initial query entry.
- // This events should be fired immediately.
+ // Initial query entry. This events should be fired immediately.
+ // A filtered one carries nothing but a counter the receiver discards,
so it is not sent at all.
if (e.updateCounter() == -1L)
- return e;
+ return e.isFiltered() ? null : e;
CacheContinuousQueryEventBuffer buf = partitionBuffer(cctx,
e.partition());
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TxDeadlockDetection.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TxDeadlockDetection.java
index 9003f5625d9..ff5d5746b5e 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TxDeadlockDetection.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TxDeadlockDetection.java
@@ -19,6 +19,7 @@ package
org.apache.ignite.internal.processors.cache.transactions;
import java.util.ArrayDeque;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -276,7 +277,7 @@ public class TxDeadlockDetection {
* @param keys Keys.
* @param txLocks Tx locks.
*/
- private void map(@Nullable Set<IgniteTxKey> keys, Map<IgniteTxKey,
List<TxLock>> txLocks) {
+ private void map(@Nullable Collection<IgniteTxKey> keys,
Map<IgniteTxKey, List<TxLock>> txLocks) {
mapTxKeys(keys, txLocks);
UUID nodeId = nodesQueue.pollFirst();
@@ -329,7 +330,7 @@ public class TxDeadlockDetection {
* @param txLocks Tx locks.
*/
@SuppressWarnings("ForLoopReplaceableByForEach")
- private void mapTxKeys(@Nullable Set<IgniteTxKey> txKeys,
Map<IgniteTxKey, List<TxLock>> txLocks) {
+ private void mapTxKeys(@Nullable Collection<IgniteTxKey> txKeys,
Map<IgniteTxKey, List<TxLock>> txLocks) {
for (Map.Entry<IgniteTxKey, List<TxLock>> e : txLocks.entrySet()) {
List<TxLock> locks = e.getValue();
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TxLocksRequest.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TxLocksRequest.java
index 9cc72fa97a7..fd8b6c0fa01 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TxLocksRequest.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TxLocksRequest.java
@@ -17,12 +17,11 @@
package org.apache.ignite.internal.processors.cache.transactions;
+import java.util.Collection;
import java.util.Set;
-import org.apache.ignite.internal.Marshalled;
import org.apache.ignite.internal.Order;
import org.apache.ignite.internal.UseBinaryMarshaller;
import org.apache.ignite.internal.processors.cache.GridCacheMessage;
-import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.internal.A;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -36,15 +35,13 @@ public class TxLocksRequest extends GridCacheMessage {
@Order(0)
long futId;
- /** Tx keys. */
- @GridToStringInclude
- @Marshalled("txKeysArr")
- Set<IgniteTxKey> txKeys;
-
- /** Array of txKeys from {@link #txKeys}. Used during marshalling and
unmarshalling. */
- @GridToStringExclude
+ /**
+ * Tx keys, deduplicated by the sender. Not a {@code Set}: the reader
would fill one while reading, and
+ * {@link IgniteTxKey#hashCode()} throws until the key's cache object is
resolved, which happens later.
+ */
@Order(1)
- IgniteTxKey[] txKeysArr;
+ @GridToStringInclude
+ Collection<IgniteTxKey> txKeys;
/**
* Default constructor.
@@ -72,7 +69,7 @@ public class TxLocksRequest extends GridCacheMessage {
}
/** */
- public Set<IgniteTxKey> txKeys() {
+ public Collection<IgniteTxKey> txKeys() {
return txKeys;
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TxLocksResponse.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TxLocksResponse.java
index f01414aecd4..3ef53c28a6e 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TxLocksResponse.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TxLocksResponse.java
@@ -18,11 +18,11 @@
package org.apache.ignite.internal.processors.cache.transactions;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import org.apache.ignite.internal.Marshalled;
import org.apache.ignite.internal.Order;
import org.apache.ignite.internal.UseBinaryMarshaller;
@@ -45,21 +45,20 @@ public class TxLocksResponse extends GridCacheMessage {
@Marshalled(keys = "nearTxKeysArr", values = "locksArr")
final Map<IgniteTxKey, List<TxLock>> nearTxKeyLocks = new HashMap<>();
- /** Remote keys involved into transactions. Doesn't include near keys. */
+ /**
+ * Remote keys involved into transactions, near keys excluded.
Deduplicated by the sender. Not a {@code Set}:
+ * the reader would fill one while reading, and {@link
IgniteTxKey#hashCode()} throws until the key's cache
+ * object is resolved, which happens later.
+ */
+ @Order(2)
@GridToStringInclude
- @Marshalled("txKeysArr")
- Set<IgniteTxKey> txKeys;
+ Collection<IgniteTxKey> txKeys;
/** Array of txKeys from {@link #nearTxKeyLocks}. Used during marshalling
and unmarshalling. */
@GridToStringExclude
@Order(1)
IgniteTxKey[] nearTxKeysArr;
- /** Array of txKeys from {@link #txKeys}. Used during marshalling and
unmarshalling. */
- @GridToStringExclude
- @Order(2)
- IgniteTxKey[] txKeysArr;
-
/** Array of locksArr from {@link #nearTxKeyLocks}. Used during
marshalling and unmarshalling. */
@GridToStringExclude
@Order(3)
@@ -114,7 +113,7 @@ public class TxLocksResponse extends GridCacheMessage {
/**
* @return Remote txKeys involved into tx.
*/
- public Set<IgniteTxKey> keys() {
+ public Collection<IgniteTxKey> keys() {
return txKeys;
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java
index ec0096383bf..4a26d329f52 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java
@@ -65,6 +65,7 @@ import
org.apache.ignite.internal.managers.collision.GridCollisionJobContextAdap
import org.apache.ignite.internal.managers.collision.GridCollisionManager;
import org.apache.ignite.internal.managers.communication.GridIoManager;
import org.apache.ignite.internal.managers.communication.GridMessageListener;
+import org.apache.ignite.internal.managers.communication.MessageMarshalling;
import org.apache.ignite.internal.managers.deployment.GridDeployment;
import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.processors.GridProcessorAdapter;
@@ -1249,10 +1250,12 @@ public class GridJobProcessor extends
GridProcessorAdapter {
boolean loc = ctx.localNodeId().equals(node.id()) &&
!ctx.config().isMarshalLocalJobs();
try {
- if (!loc)
- req.unmarshal(marsh,
U.resolveClassLoader(dep.classLoader(), ctx.config()));
+ // The job payload waits for this point: only now is
there a deployment to unmarshal it with.
+ if (!loc) {
+ MessageMarshalling.unmarshal(req, ctx, null,
+ U.resolveClassLoader(dep.classLoader(),
ctx.config()));
+ }
- // Note that we unmarshal session/job attributes here
with proper class loader.
GridTaskSessionImpl taskSes =
ctx.session().createTaskSession(
req.sessionId(),
node.id(),
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/message/QueryEntityMessage.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/message/QueryEntityMessage.java
index e66f2a17df5..27ad8c0eb5a 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/message/QueryEntityMessage.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/message/QueryEntityMessage.java
@@ -23,19 +23,18 @@ import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cache.QueryEntity;
-import org.apache.ignite.internal.MarshallableMessage;
+import org.apache.ignite.internal.Marshalled;
import org.apache.ignite.internal.Order;
import org.apache.ignite.internal.cache.query.QueryIndexMessage;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.plugin.extensions.communication.Message;
/**
* Message for {@link QueryEntity} transfer.
*/
-public class QueryEntityMessage implements MarshallableMessage {
+public class QueryEntityMessage implements Message {
/** Key type. */
@Order(0)
String keyType;
@@ -80,6 +79,7 @@ public class QueryEntityMessage implements
MarshallableMessage {
Set<String> notNullFields;
/** Fields default values. */
+ @Marshalled("dfltFieldValuesBytes")
Map<String, Object> dfltFieldValues;
/** Serialized form of {@link #dfltFieldValues}. */
@@ -117,7 +117,10 @@ public class QueryEntityMessage implements
MarshallableMessage {
tableName = qryEntity.getTableName();
notNullFields = qryEntity.getNotNullFields();
- dfltFieldValues = qryEntity.getDefaultFieldValues();
+
+ if (!F.isEmpty(qryEntity.getDefaultFieldValues()))
+ dfltFieldValues = qryEntity.getDefaultFieldValues();
+
fieldsPrecision = qryEntity.getFieldsPrecision();
fieldsScale = qryEntity.getFieldsScale();
}
@@ -143,19 +146,4 @@ public class QueryEntityMessage implements
MarshallableMessage {
.setFieldsPrecision(fieldsPrecision)
.setFieldsScale(fieldsScale);
}
-
- /** {@inheritDoc} */
- @Override public void marshal(Marshaller marsh) throws
IgniteCheckedException {
- if (!F.isEmpty(dfltFieldValues) && dfltFieldValuesBytes == null)
- dfltFieldValuesBytes = U.marshal(marsh, dfltFieldValues);
- }
-
- /** {@inheritDoc} */
- @Override public void unmarshal(Marshaller marsh, ClassLoader clsLdr)
throws IgniteCheckedException {
- if (!F.isEmpty(dfltFieldValuesBytes)) {
- dfltFieldValues = U.unmarshal(marsh, dfltFieldValuesBytes, clsLdr);
-
- dfltFieldValuesBytes = null;
- }
- }
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/LazyServiceConfigurationMessage.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/LazyServiceConfigurationMessage.java
index 7c2fddac971..d3526af96ab 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/LazyServiceConfigurationMessage.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/LazyServiceConfigurationMessage.java
@@ -17,18 +17,15 @@
package org.apache.ignite.internal.processors.service;
-import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.internal.MarshallableMessage;
+import org.apache.ignite.internal.Marshalled;
import org.apache.ignite.internal.Order;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
-import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.S;
-import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.extensions.communication.MessageFactory;
/** Message for {@link LazyServiceConfiguration}. */
-public class LazyServiceConfigurationMessage implements MarshallableMessage {
+public class LazyServiceConfigurationMessage implements Message {
/** Service name. */
@Order(0)
String name;
@@ -46,7 +43,8 @@ public class LazyServiceConfigurationMessage implements
MarshallableMessage {
String cacheName;
/** Affinity key. */
- private Object affKey;
+ @Marshalled("affKeyBytes")
+ Object affKey;
/** Serialized {@link #affKey}. */
@Order(4)
@@ -122,21 +120,6 @@ public class LazyServiceConfigurationMessage implements
MarshallableMessage {
.platformMtdNames(platformMtdNames);
}
- /** {@inheritDoc} */
- @Override public void marshal(Marshaller marsh) throws
IgniteCheckedException {
- if (affKey != null && affKeyBytes == null)
- affKeyBytes = U.marshal(marsh, affKey);
- }
-
- /** {@inheritDoc} */
- @Override public void unmarshal(Marshaller marsh, ClassLoader clsLdr)
throws IgniteCheckedException {
- if (!F.isEmpty(affKeyBytes)) {
- affKey = U.unmarshal(marsh, affKeyBytes, clsLdr);
-
- affKeyBytes = null;
- }
- }
-
/** {@inheritDoc} */
@Override public String toString() {
return S.toString(LazyServiceConfigurationMessage.class, this);
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java
index fe419cefaf6..31e7b65bd55 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java
@@ -1416,8 +1416,6 @@ public class GridTaskWorker<T, R> extends GridWorker
implements GridTimeoutObjec
if (loc)
ctx.job().processJobExecuteRequest(ctx.discovery().localNode(), req);
else {
- req.marshal(marsh);
-
byte plc;
if (internal)
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareDestroyedCacheTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareDestroyedCacheTest.java
index 6ebace2cd04..8d6f7426966 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareDestroyedCacheTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareDestroyedCacheTest.java
@@ -32,7 +32,7 @@ import org.apache.ignite.failure.FailureHandler;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.TestRecordingCommunicationSpi;
-import
org.apache.ignite.internal.processors.cache.CacheObjectNotResolvedException;
+import org.apache.ignite.internal.processors.cache.CacheInvalidStateException;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.ListeningTestLogger;
@@ -47,10 +47,10 @@ import static
org.apache.ignite.transactions.TransactionConcurrency.OPTIMISTIC;
import static
org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
/**
- * Checks the receiver's behaviour when a tx prepare of a destroyed cache
cannot be unmarshalled: without the cache
- * there is no cache object context, so the keys stay unresolved and
rebuilding the DHT-version map hashes such a key
- * and throws {@link CacheObjectNotResolvedException}. The receiver must
convert it to an error response for the
- * sender and stay alive instead of going down through the failure handler.
+ * Checks the receiver's behaviour when a tx prepare arrives for a cache that
has been destroyed: the entry has no
+ * cache context to initialize against, so the prepare fails with {@link
CacheInvalidStateException}. The receiver
+ * must convert it to an error response for the sender and stay alive instead
of going down through the failure
+ * handler.
*/
public class GridNearTxPrepareDestroyedCacheTest extends
GridCommonAbstractTest {
/** */
@@ -124,8 +124,8 @@ public class GridNearTxPrepareDestroyedCacheTest extends
GridCommonAbstractTest
IgniteInternalFuture<?> txFut = GridTestUtils.runAsync(() -> {
try (Transaction tx = near.transactions().txStart(OPTIMISTIC,
READ_COMMITTED)) {
- // A transform entry adds its key to the prepare's DHT-version
map, so the receiver hashes the key
- // while rebuilding the map.
+ // A transform entry is what puts a key into the prepare's
DHT-version keys, so the request
+ // delivered below carries one.
cache.invoke(key, new SetValueProcessor());
tx.commit();
@@ -163,13 +163,13 @@ public class GridNearTxPrepareDestroyedCacheTest extends
GridCommonAbstractTest
Throwable err = ((GridNearTxPrepareResponse)resps.get(0)).error();
- assertNotNull("The response must carry the unmarshalling error.", err);
+ assertNotNull("The response must carry the error of the destroyed
cache.", err);
- CacheObjectNotResolvedException cause = X.cause(err,
CacheObjectNotResolvedException.class);
+ CacheInvalidStateException cause = X.cause(err,
CacheInvalidStateException.class);
assertNotNull("Unexpected response error: " +
X.getFullStackTrace(err), cause);
- assertTrue(cause.getMessage(), cause.getMessage().contains("Cache
object is not deserialized"));
+ assertTrue(cause.getMessage(), cause.getMessage().contains("cache is
stopped"));
// The transaction had been rolled back by the cache stop before the
prepare was delivered, so the arrived
// error response finds no future to complete.
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryImmutableEntryTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryImmutableEntryTest.java
index ebed66410dd..470d0d0363c 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryImmutableEntryTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryImmutableEntryTest.java
@@ -38,7 +38,6 @@ import
org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheObjectImpl;
import org.apache.ignite.internal.processors.cache.KeyCacheObjectImpl;
import org.apache.ignite.internal.util.nio.MessageSerialization;
-import org.apache.ignite.internal.util.typedef.internal.U;
import
org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.Test;
@@ -136,7 +135,50 @@ public class IgniteCacheContinuousQueryImmutableEntryTest
extends GridCommonAbst
*/
@Test
public void testCacheContinuousQueryEntrySerialization() throws Exception {
- CacheContinuousQueryEntry e0 = new CacheContinuousQueryEntry(
+ CacheContinuousQueryEntry e0 = entryWithData();
+
+ e0.markFiltered();
+
+ CacheContinuousQueryEntry sent = sendForm(e0);
+ CacheContinuousQueryEntry e1 = roundTrip(sent);
+
+ assertEquals(sent.cacheId(), e1.cacheId());
+ assertEquals(sent.eventType(), e1.eventType());
+ assertEquals(sent.isFiltered(), e1.isFiltered());
+ assertEquals(sent.isBackup(), e1.isBackup());
+ assertEquals(sent.isKeepBinary(), e1.isKeepBinary());
+ assertEquals(sent.partition(), e1.partition());
+ assertEquals(sent.updateCounter(), e1.updateCounter());
+ }
+
+ /** A filtered entry gives up its data at the buffer, the only door
between an update and the wire. */
+ @Test
+ public void testFilteredEntryLosesDataInBuffer() {
+ CacheContinuousQueryEntry e0 = entryWithData();
+
+ e0.markFiltered();
+
+ CacheContinuousQueryEntry sent = sendForm(e0);
+
+ assertTrue(sent.isFiltered());
+
+ assertNull(sent.key());
+ assertNull(sent.newValue());
+ assertNull(sent.oldValue());
+
+ assertNotNull(e0.key());
+ assertNotNull(e0.newValue());
+ assertNotNull(e0.oldValue());
+ }
+
+ /** @return Entry the buffer hands back to be sent: a counter below the
batch start goes out as is. */
+ private CacheContinuousQueryEntry sendForm(CacheContinuousQueryEntry e) {
+ return (CacheContinuousQueryEntry)new
CacheContinuousQueryEventBuffer(log).processEntry(e, false);
+ }
+
+ /** @return Entry carrying a key and both values. */
+ private CacheContinuousQueryEntry entryWithData() {
+ return new CacheContinuousQueryEntry(
1,
EventType.UPDATED,
new KeyCacheObjectImpl(1, new byte[] {0, 0, 0, 1}, 1),
@@ -144,55 +186,36 @@ public class IgniteCacheContinuousQueryImmutableEntryTest
extends GridCommonAbst
new CacheObjectImpl(2, new byte[] {0, 0, 0, 3}),
true,
1,
- 1L,
+ 0L,
new AffinityTopologyVersion(1L),
(byte)0);
+ }
- e0.markFiltered();
-
+ /** @return Entry read back from the bytes {@code e} is written to. */
+ private CacheContinuousQueryEntry roundTrip(CacheContinuousQueryEntry e)
throws Exception {
IgniteMessageFactoryImpl msgFactory =
new IgniteMessageFactoryImpl(new MessageFactoryProvider[]{new
CoreMessagesProvider(jdk(), jdk())});
ByteBuffer buf = ByteBuffer.allocate(4096);
DirectMessageWriter writer = new DirectMessageWriter(msgFactory);
- assertNotNull("Serializer not found for message type " +
e0.directType(), msgFactory.serializer(e0.directType()));
+ assertNotNull("Serializer not found for message type " +
e.directType(), msgFactory.serializer(e.directType()));
writer.setBuffer(buf);
- // The marshal phase run by the send path before serialization: must
not touch the wire fields of
- // a filtered entry.
- e0.marshal(jdk());
-
// Skip write class header.
writer.onHeaderWritten();
- MessageSerialization.writeTo(msgFactory, e0, writer);
+ MessageSerialization.writeTo(msgFactory, e, writer);
- CacheContinuousQueryEntry e1 = new CacheContinuousQueryEntry();
+ CacheContinuousQueryEntry res = new CacheContinuousQueryEntry();
- final DirectMessageReader reader = new DirectMessageReader(msgFactory,
null);
+ DirectMessageReader reader = new DirectMessageReader(msgFactory, null);
reader.setBuffer(ByteBuffer.wrap(buf.array()));
- MessageSerialization.readFrom(msgFactory, e1, reader);
-
- e1.unmarshal(jdk(), U.gridClassLoader());
-
- assertEquals(e0.cacheId(), e1.cacheId());
- assertEquals(e0.eventType(), e1.eventType());
- assertEquals(e0.isFiltered(), e1.isFiltered());
- assertEquals(e0.isBackup(), e1.isBackup());
- assertEquals(e0.isKeepBinary(), e1.isKeepBinary());
- assertEquals(e0.partition(), e1.partition());
- assertEquals(e0.updateCounter(), e1.updateCounter());
+ MessageSerialization.readFrom(msgFactory, res, reader);
- // Key and value shouldn't be serialized in case an event is filtered.
- assertNull(e1.key());
- assertNotNull(e0.key());
- assertNull(e1.oldValue());
- assertNotNull(e0.oldValue());
- assertNull(e1.newValue());
- assertNotNull(e0.newValue());
+ return res;
}
/**
