shishkovilja commented on code in PR #13095:
URL: https://github.com/apache/ignite/pull/13095#discussion_r3572053259
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java:
##########
@@ -98,14 +100,14 @@ public class GridDhtAtomicUpdateRequest extends
GridDhtAtomicAbstractUpdateReque
boolean forceTransformBackups;
/** Entry processors. */
- private List<EntryProcessor<Object, Object, Object>> entryProcessors;
+ List<EntryProcessor<Object, Object, Object>> entryProcessors;
Review Comment:
It seems, that change is unnecessary.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/MetadataUpdateProposedMessage.java:
##########
@@ -73,13 +71,14 @@
* it gets blocked until {@link MetadataUpdateAcceptedMessage} arrives with
<b>accepted version</b>
* equals to <b>pending version</b> of this metadata to the moment when is was
initially read by the thread.
*/
-public final class MetadataUpdateProposedMessage extends
DiscoveryCustomMessage implements MarshallableMessage {
+public final class MetadataUpdateProposedMessage extends
DiscoveryCustomMessage implements Message {
Review Comment:
```suggestion
public final class MetadataUpdateProposedMessage extends
DiscoveryCustomMessage {
```
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicFullUpdateRequest.java:
##########
@@ -62,7 +65,7 @@ public class GridNearAtomicFullUpdateRequest extends
GridNearAtomicAbstractUpdat
List<CacheObject> vals;
/** Entry processors. */
- private List<EntryProcessor<Object, Object, Object>> entryProcessors;
+ List<EntryProcessor<Object, Object, Object>> entryProcessors;
Review Comment:
Unneccessary change.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java:
##########
@@ -98,14 +100,14 @@ public class GridDhtAtomicUpdateRequest extends
GridDhtAtomicAbstractUpdateReque
boolean forceTransformBackups;
/** Entry processors. */
- private List<EntryProcessor<Object, Object, Object>> entryProcessors;
+ List<EntryProcessor<Object, Object, Object>> entryProcessors;
/** Entry processors bytes. */
@Order(12)
List<byte[]> entryProcessorsBytes;
/** Near entry processors. */
- private List<EntryProcessor<Object, Object, Object>> nearEntryProcessors;
+ List<EntryProcessor<Object, Object, Object>> nearEntryProcessors;
Review Comment:
It seems, that change is unnecessary.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsSingleMessage.java:
##########
@@ -23,22 +23,23 @@
import java.util.Map;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.internal.Compress;
+import org.apache.ignite.internal.MarshallableMessage;
import org.apache.ignite.internal.Order;
import org.apache.ignite.internal.managers.communication.ErrorMessage;
-import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import
org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState;
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.marshaller.Marshaller;
import org.jetbrains.annotations.Nullable;
/**
* Information about partitions of a single node. <br>
*
* Sent in response to {@link GridDhtPartitionsSingleRequest} and during
processing partitions exchange future.
*/
-public class GridDhtPartitionsSingleMessage extends
GridDhtPartitionsAbstractMessage {
+public class GridDhtPartitionsSingleMessage extends
GridDhtPartitionsAbstractMessage implements MarshallableMessage {
Review Comment:
The same situation as with the `GridDhtPartitionsFullMessage`:
`GridDhtPartitionsSingleMessage` is no marshallable message.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java:
##########
@@ -1076,7 +1073,7 @@ private void notifyCallback0(UUID nodeId,
}
}
- e.unmarshal(cctx, ldr);
+ MessageMarshaller.unmarshal(ctx.messageFactory(), e, ctx,
coctx, depMgr.globalLoader());
Review Comment:
Why we keep unmarshalling, will it occur twice?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java:
##########
@@ -418,11 +418,11 @@ public void onEntryUpdated(
oldVal = (CacheObject)cctx.unwrapTemporary(oldVal);
if (oldVal != null)
- oldVal.finishUnmarshal(cctx.cacheObjectContext(),
cctx.deploy().globalLoader());
+ oldVal.unmarshal(cctx.cacheObjectContext(),
cctx.deploy().globalLoader());
}
if (newVal != null)
- newVal.finishUnmarshal(cctx.cacheObjectContext(),
cctx.deploy().globalLoader());
+ newVal.unmarshal(cctx.cacheObjectContext(),
cctx.deploy().globalLoader());
Review Comment:
Can unmarshalling occur twice?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java:
##########
@@ -418,11 +418,11 @@ public void onEntryUpdated(
oldVal = (CacheObject)cctx.unwrapTemporary(oldVal);
if (oldVal != null)
- oldVal.finishUnmarshal(cctx.cacheObjectContext(),
cctx.deploy().globalLoader());
+ oldVal.unmarshal(cctx.cacheObjectContext(),
cctx.deploy().globalLoader());
Review Comment:
Can unmarshalling occur twice?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheObjectNotResolvedException.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * 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;
+
+/**
+ * Thrown by {@link KeyCacheObjectImpl#hashCode()} when the object has not
been deserialized yet,
+ * which happens when the owning cache has been removed before {@code
unmarshal} could complete.
+ */
+public class CacheObjectNotResolvedException extends RuntimeException {
Review Comment:
Does we have any test for this exception?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TxLocksResponse.java:
##########
@@ -96,7 +97,7 @@ public Map<IgniteTxKey, List<TxLock>> txLocks() {
* @return Lock list for given tx key.
*/
public List<TxLock> txLocks(IgniteTxKey txKey) {
- return nearTxKeyLocks.get(txKey);
Review Comment:
Unnecessary change.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryInfo.java:
##########
@@ -201,26 +191,8 @@ public int marshalledSize(CacheObjectContext ctx) throws
IgniteCheckedException
return SIZE_OVERHEAD + size;
}
- /**
- * @param ctx Cache context.
- * @throws IgniteCheckedException In case of error.
- */
- public void marshal(GridCacheContext ctx) throws IgniteCheckedException {
- marshal(ctx.cacheObjectContext());
- }
-
- /**
- * @param ctx Cache context.
- * @throws IgniteCheckedException In case of error.
- */
- public void marshal(CacheObjectContext ctx) throws IgniteCheckedException {
- assert key != null;
-
- key.prepareMarshal(ctx);
-
- if (val != null)
- val.prepareMarshal(ctx);
-
+ /** {@inheritDoc} */
+ @Override public void marshal(Marshaller marsh) throws
IgniteCheckedException {
Review Comment:
Not a marshallable, in fact.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java:
##########
@@ -1931,12 +1931,12 @@ private void submit(
else {
try {
for (DataStreamerEntry e : entries) {
- e.getKey().prepareMarshal(cacheObjCtx);
+ e.getKey().marshal(cacheObjCtx);
Review Comment:
Possible duplicated marshalling?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java:
##########
@@ -3876,7 +3877,7 @@ else if (exchCtx.events().hasServerLeft())
else if (forceAffReassignment)
msg.idealAffinityDiff(idealAffDiff);
- msg.prepareMarshal(cctx);
+
GridCacheMessageDeployer.deploy(cctx.kernalContext().messageFactory(), msg,
cctx);
Review Comment:
Why `deploy`, when call of `marshall` is necessary?
As I see, `GridDhtPartitionsFullMessage` is not a deployable message and
even not a marshallable one.
We can slightly refactor `GridDhtPartitionsFullMessage` (may be in a
separate PR), eg.:
1. `marshall` -> `copyLocalPartitions`
2. `unmarshall` -> `deduplicatePartitionsIfNecessary`
But, it is necessary to figure out, where we should call `unmarshall`.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerUpdateJob.java:
##########
@@ -117,14 +117,14 @@ class DataStreamerUpdateJob implements
GridPlainCallable<Object> {
final GridCacheContext cctx = cache.context();
for (DataStreamerEntry e : col) {
- e.getKey().finishUnmarshal(cctx.cacheObjectContext(),
cctx.deploy().globalLoader());
+ e.getKey().unmarshal(cctx.cacheObjectContext(),
cctx.deploy().globalLoader());
Review Comment:
Possible duplicated unmarshalling?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageImpl.java:
##########
@@ -1039,7 +1039,7 @@ private GridFutureAdapter<Boolean> startCas(String key,
@Nullable Serializable e
DistributedMetaStorageCasMessage msg = new
DistributedMetaStorageCasMessage(reqId, key, expVal, newVal);
- msg.prepareMarshal(marshaller);
+ msg.marshal(marshaller);
Review Comment:
Duplicated marshalling?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java:
##########
@@ -1250,7 +1250,7 @@ public void processJobExecuteRequest(ClusterNode node,
final GridJobExecuteReque
try {
if (!loc)
- req.finishUnmarshal(marsh,
U.resolveClassLoader(dep.classLoader(), ctx.config()));
+ req.unmarshal(marsh,
U.resolveClassLoader(dep.classLoader(), ctx.config()));
Review Comment:
Possible duplicated unmarshalling?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java:
##########
@@ -1416,7 +1416,7 @@ private void sendRequest(ComputeJobResult res) {
if (loc)
ctx.job().processJobExecuteRequest(ctx.discovery().localNode(), req);
else {
- req.prepareMarshal(marsh);
+ req.marshal(marsh);
Review Comment:
`GridJobExecuteRequest` is not marshallable, but it can confuse, that we
have `marshal` for `MarshallableMessage` and non marshallable one.
##########
modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java:
##########
@@ -1319,7 +1320,8 @@ else if (e instanceof IgniteCheckedException)
}
/** */
- private void initMessageFactory() throws IgniteCheckedException {
+ @TestOnly
Review Comment:
Why TestOnly?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java:
##########
@@ -995,13 +995,18 @@ private long allocateForTree() throws
IgniteCheckedException {
assert info.ttl() == TTL_ETERNAL : info.ttl();
- batch.add(new DataRowCacheAware(info.key(),
- info.value(),
- info.version(),
- part.id(),
- info.expireTime(),
- info.cacheId(),
- grp.storeCacheIdInDataPage()));
+ try {
+ batch.add(new DataRowCacheAware(info.key(),
+ info.value(),
+ info.version(),
+ part.id(),
+ info.expireTime(),
+ info.cacheId(),
+ grp.storeCacheIdInDataPage()));
+ }
+ catch (IllegalStateException th) {
+ assert ctx.cacheContext(grp.groupId()) == null; // Ignoring
removed cache entries.
+ }
Review Comment:
Why do we need this change? Any test fails or something else?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareRequest.java:
##########
@@ -296,6 +293,25 @@ private boolean isFlag(int mask) {
return U.safeAbs(version().hashCode());
}
+ /** {@inheritDoc} */
Review Comment:
It is not marshalling in fact. I suggest to use more specific method and
call it where it is necessary.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TxEntryValueHolder.java:
##########
@@ -119,25 +116,6 @@ public boolean hasReadValue() {
return hasReadVal;
}
- /**
- * @param ctx Cache context.
- * @throws IgniteCheckedException If marshaling failed.
- */
- public void marshal(GridCacheContext<?, ?> ctx) throws
IgniteCheckedException {
- if (hasWriteVal && val != null)
- val.prepareMarshal(ctx.cacheObjectContext());
- }
-
- /**
- * @param ctx Cache context.
- * @param ldr Class loader.
- * @throws IgniteCheckedException If unmarshalling failed.
- */
- public void unmarshal(CacheObjectValueContext ctx, ClassLoader ldr) throws
IgniteCheckedException {
- if (hasWriteVal && val != null)
Review Comment:
We have lost some marshalling logic here.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java:
##########
@@ -932,13 +934,8 @@ protected CacheEntryEventFilter getEventFilter0() {
*/
private void prepareEntry(GridCacheContext cctx, UUID nodeId,
CacheContinuousQueryEntry entry)
throws IgniteCheckedException {
- if (cctx.kernalContext().config().isPeerClassLoadingEnabled() &&
cctx.discovery().node(nodeId) != null) {
Review Comment:
Why we remove marshalling?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java:
##########
@@ -484,7 +484,7 @@ public void onEntryExpired(GridCacheEntryEx e,
KeyCacheObject key, CacheObject o
oldVal = (CacheObject)cctx.unwrapTemporary(oldVal);
if (oldVal != null)
- oldVal.finishUnmarshal(cctx.cacheObjectContext(),
cctx.deploy().globalLoader());
+ oldVal.unmarshal(cctx.cacheObjectContext(),
cctx.deploy().globalLoader());
Review Comment:
Can unmarshalling occur twice?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java:
##########
@@ -430,12 +432,21 @@ public IgniteInternalFuture<GridNearTxPrepareResponse>
prepareNearTxLocal(
IgniteTxEntry firstEntry;
try {
- IgniteTxEntry firstWrite = unmarshal(req.writes());
- IgniteTxEntry firstRead = unmarshal(req.reads());
+ IgniteTxEntry firstWrite = initialize(req.writes(),
req.topologyVersion());
+ IgniteTxEntry firstRead = initialize(req.reads(),
req.topologyVersion());
firstEntry = firstWrite != null ? firstWrite : firstRead;
}
catch (IgniteCheckedException e) {
+ try {
+ req.onClassError(e);
Review Comment:
Error handling was changed. Can you explain, why do we need this? Some tests
fails or something else?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryInfo.java:
##########
@@ -231,30 +203,8 @@ public void marshal(CacheObjectContext ctx) throws
IgniteCheckedException {
}
}
- /**
- * Unmarshalls entry.
- *
- * @param ctx Cache context.
- * @param clsLdr Class loader.
- * @throws IgniteCheckedException If unmarshalling failed.
- */
- public void unmarshal(GridCacheContext ctx, ClassLoader clsLdr) throws
IgniteCheckedException {
- unmarshal(ctx.cacheObjectContext(), clsLdr);
- }
-
- /**
- * Unmarshalls entry.
- *
- * @param ctx Cache context.
- * @param clsLdr Class loader.
- * @throws IgniteCheckedException If unmarshalling failed.
- */
- public void unmarshal(CacheObjectContext ctx, ClassLoader clsLdr) throws
IgniteCheckedException {
- key.finishUnmarshal(ctx, clsLdr);
-
- if (val != null)
- val.finishUnmarshal(ctx, clsLdr);
-
+ /** {@inheritDoc} */
+ @Override public void unmarshal(Marshaller marsh, ClassLoader clsLdr)
throws IgniteCheckedException {
Review Comment:
Not a marshallable, in fact.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java:
##########
@@ -1931,12 +1931,12 @@ private void submit(
else {
try {
for (DataStreamerEntry e : entries) {
- e.getKey().prepareMarshal(cacheObjCtx);
+ e.getKey().marshal(cacheObjCtx);
CacheObject val = e.getValue();
if (val != null)
- val.prepareMarshal(cacheObjCtx);
+ val.marshal(cacheObjCtx);
Review Comment:
Possible duplicated marshalling?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/StoredCacheData.java:
##########
@@ -45,21 +43,23 @@
* <p>
* All changes must be made with the respect of RU rules.
*/
-public class StoredCacheData implements Serializable, CdcCacheEvent,
MarshallableMessage {
+public class StoredCacheData implements Serializable, CdcCacheEvent, Message {
/** */
private static final long serialVersionUID = 0L;
/** Cache configuration. */
@GridToStringInclude
- private CacheConfiguration<?, ?> ccfg;
+ @Marshalled("ccfgBytes")
+ CacheConfiguration<?, ?> ccfg;
/** Serialized {@link #ccfg}. */
@Order(0)
transient byte[] ccfgBytes;
/** Query entities. */
@GridToStringInclude
- private Collection<QueryEntity> qryEntities;
+ @Marshalled("qryEntitiesBytes")
+ Collection<QueryEntity> qryEntities;
Review Comment:
Let's use `QueryEntityMessage`.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java:
##########
@@ -2253,7 +2253,7 @@ protected static class IsolatedUpdater implements
StreamReceiver<KeyCacheObject,
cctx.shared().database().checkpointReadLock();
try {
- e.getKey().finishUnmarshal(cctx.cacheObjectContext(),
cctx.deploy().globalLoader());
+ e.getKey().unmarshal(cctx.cacheObjectContext(),
cctx.deploy().globalLoader());
Review Comment:
Possible duplicated unmarshalling?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerUpdateJob.java:
##########
@@ -117,14 +117,14 @@ class DataStreamerUpdateJob implements
GridPlainCallable<Object> {
final GridCacheContext cctx = cache.context();
for (DataStreamerEntry e : col) {
- e.getKey().finishUnmarshal(cctx.cacheObjectContext(),
cctx.deploy().globalLoader());
+ e.getKey().unmarshal(cctx.cacheObjectContext(),
cctx.deploy().globalLoader());
CacheObject val = e.getValue();
if (val != null) {
checkSecurityPermission(SecurityPermission.CACHE_PUT);
- val.finishUnmarshal(cctx.cacheObjectContext(),
cctx.deploy().globalLoader());
+ val.unmarshal(cctx.cacheObjectContext(),
cctx.deploy().globalLoader());
Review Comment:
Possible duplicated unmarshalling?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageImpl.java:
##########
@@ -1281,7 +1281,7 @@ private void completeCas(
Serializable oldVal = bridge.read(msg.key());
- msg.finishUnmarshal(marshaller);
+ msg.unmarshal(marshaller);
Review Comment:
Duplicated unmarshalling?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
