This is an automated email from the ASF dual-hosted git repository.
sergey-chugunov-1985 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 b8a009ececa IGNITE-28994 GridDhtPartitionsFullMessage might not
implement undesirable SelfMarshallingMessage (#13493)
b8a009ececa is described below
commit b8a009ececa3ec24fdb64b210684c8f5a111c4d6
Author: Vladimir Steshin <[email protected]>
AuthorDate: Thu Aug 20 13:05:00 2026 +0300
IGNITE-28994 GridDhtPartitionsFullMessage might not implement undesirable
SelfMarshallingMessage (#13493)
---
.../query/calcite/message/QueryStartRequest.java | 2 +-
.../calcite/metadata/FragmentDescription.java | 12 +++++-----
.../cache/GridCachePartitionExchangeManager.java | 26 +++++-----------------
.../preloader/GridDhtPartitionsExchangeFuture.java | 9 ++++++--
.../preloader/GridDhtPartitionsFullMessage.java | 17 +++++++++-----
5 files changed, 30 insertions(+), 36 deletions(-)
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 58c40688c23..75150160722 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
@@ -146,7 +146,7 @@ public class QueryStartRequest implements
DeferredUnmarshalMessage, ExecutionCon
* @return Fragment description.
*/
public FragmentDescription fragmentDescription() {
- return fragmentDesc.receivedFragment();
+ return fragmentDesc.received();
}
/**
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/FragmentDescription.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/FragmentDescription.java
index 9794e730f28..0a7fd365844 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/FragmentDescription.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/FragmentDescription.java
@@ -25,11 +25,11 @@ import
org.apache.ignite.plugin.extensions.communication.Message;
import org.jetbrains.annotations.Nullable;
/**
- * Query fragment description. Has to be prepared to send to another node and
to restore after receiving from another
- * node.
+ * Query fragment description. <br>
+ * Has to be prepared to send to another node and restored after receiving
from another node.
*
* @see #preparedToSend()
- * @see #receivedFragment()
+ * @see #received()
*/
public class FragmentDescription implements Message {
/** */
@@ -48,7 +48,7 @@ public class FragmentDescription implements Message {
@Order(3)
@Nullable ColocationGroup target;
- /** Transient flag of {@link #receivedFragment()}-once-invoked. */
+ /** Transient flag of {@link #received()}-once-invoked. */
boolean received;
/** */
@@ -77,8 +77,8 @@ public class FragmentDescription implements Message {
return this;
}
- /** Properly unwraps this fragment description after receiving from
another node. */
- public FragmentDescription receivedFragment() {
+ /** Properly unwraps fragment description after receiving from another
node. */
+ public FragmentDescription received() {
if (!received) {
if (target != null)
target.afterReceive();
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
index 50630e297f0..8494d8f57db 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
@@ -425,6 +425,8 @@ public class GridCachePartitionExchangeManager<K, V>
extends GridCacheSharedMana
cctx.io().addCacheHandler(GridDhtPartitionsFullMessage.class,
new MessageHandler<GridDhtPartitionsFullMessage>() {
@Override public void onMessage(ClusterNode node,
GridDhtPartitionsFullMessage msg) {
+ msg.received();
+
if (msg.exchangeId() == null) {
GridDhtPartitionsExchangeFuture curExchange =
lastTopologyFuture();
@@ -1311,27 +1313,7 @@ public class GridCachePartitionExchangeManager<K, V>
extends GridCacheSharedMana
}
/**
- * Creates partitions full message for all cache groups.
- *
- * @param exchId Non-null exchange ID if message is created for exchange.
- * @param lastVer Last version.
- * @param partHistSuppliers Partition history suppliers map.
- * @param partsToReload Partitions to reload map.
- * @return Message.
- */
- public GridDhtPartitionsFullMessage createPartitionsFullMessage(
- @Nullable final GridDhtPartitionExchangeId exchId,
- @Nullable GridCacheVersion lastVer,
- Map<UUID, Map<GroupPartitionIdPair, Long>> partHistSuppliers,
- @Nullable Map<UUID, Map<Integer, Set<Integer>>> partsToReload
- ) {
- Collection<CacheGroupContext> grps = cctx.cache().cacheGroups();
-
- return createPartitionsFullMessage(exchId, lastVer, partHistSuppliers,
partsToReload, grps);
- }
-
- /**
- * Creates partitions full message for selected cache groups.
+ * Creates partitions full message for selected cache groups and prepares
it to send to another node.
*
* @param exchId Non-null exchange ID if message is created for exchange.
* @param lastVer Last version.
@@ -1406,6 +1388,8 @@ public class GridCachePartitionExchangeManager<K, V>
extends GridCacheSharedMana
if (!partsSizes.isEmpty())
m.partitionSizes(partsSizes);
+ m.prepareToSend();
+
return m;
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
index fe254d14b72..2cc8a517218 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
@@ -286,6 +286,7 @@ public class GridDhtPartitionsExchangeFuture extends
GridDhtTopologyFutureAdapte
/**
* Centralized affinity assignment required. Activated for node left of
failed. For this mode crd will send full
* partitions maps to nodes using discovery (ring) instead of
communication.
+ * TODO : seems is always false,
https://issues.apache.org/jira/browse/IGNITE-28997
*/
private boolean centralizedAff;
@@ -2110,7 +2111,9 @@ public class GridDhtPartitionsExchangeFuture extends
GridDhtTopologyFutureAdapte
exchangeId(),
last != null ? last : cctx.versions().last(),
partHistSuppliers,
- partsToReload);
+ partsToReload,
+ cctx.cache().cacheGroups()
+ );
if (stateChangeExchange() && !F.isEmpty(exchangeGlobalExceptions))
m.setErrorsMap(exchangeGlobalExceptions);
@@ -3202,6 +3205,7 @@ public class GridDhtPartitionsExchangeFuture extends
GridDhtTopologyFutureAdapte
/**
* @param fut Affinity future.
+ * TODO : Seems isn't used
https://issues.apache.org/jira/browse/IGNITE-28997
*/
private void onAffinityInitialized(IgniteInternalFuture<Map<Integer,
Map<Integer, List<UUID>>>> fut) {
try {
@@ -3831,6 +3835,7 @@ public class GridDhtPartitionsExchangeFuture extends
GridDhtTopologyFutureAdapte
state = ExchangeLocalState.DONE;
}
+ // TODO : seems is always false,
https://issues.apache.org/jira/browse/IGNITE-28997
if (centralizedAff) {
assert !exchCtx.mergeExchanges();
@@ -4746,7 +4751,7 @@ public class GridDhtPartitionsExchangeFuture extends
GridDhtTopologyFutureAdapte
cctx.affinity().onExchangeChangeAffinityMessage(GridDhtPartitionsExchangeFuture.this,
msg);
- GridDhtPartitionsFullMessage partsMsg =
msg.partitionsMessage();
+ GridDhtPartitionsFullMessage partsMsg =
msg.partitionsMessage().received();
IgniteCheckedException err =
!F.isEmpty(partsMsg.getErrorsMap()) ?
new IgniteCheckedException("Cluster state change
failed.") : null;
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsFullMessage.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsFullMessage.java
index 618cf37eadb..e18b894f741 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsFullMessage.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsFullMessage.java
@@ -28,7 +28,6 @@ import java.util.stream.IntStream;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.Compress;
import org.apache.ignite.internal.Order;
-import org.apache.ignite.internal.SelfMarshallingMessage;
import org.apache.ignite.internal.managers.discovery.GridDiscoveryManager;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheAffinityChangeMessage;
@@ -46,8 +45,12 @@ import org.jetbrains.annotations.Nullable;
* Information about partitions of all nodes in topology. <br> Is sent by
topology coordinator: when all {@link
* GridDhtPartitionsSingleMessage}s were received. <br> May be also compacted
as part of {@link
* CacheAffinityChangeMessage} for node left or failed case.<br>
+ * Has to be prepared to send to another node and restored after receiving
from another node.
+ *
+ * @see #prepareToSend()
+ * @see #received()
*/
-public class GridDhtPartitionsFullMessage extends
GridDhtPartitionsAbstractMessage implements SelfMarshallingMessage {
+public class GridDhtPartitionsFullMessage extends
GridDhtPartitionsAbstractMessage {
/** */
private static final byte REBALANCED_FLAG_MASK = 0x01;
@@ -385,8 +388,8 @@ public class GridDhtPartitionsFullMessage extends
GridDhtPartitionsAbstractMessa
flags = rebalanced ? (byte)(flags | REBALANCED_FLAG_MASK) :
(byte)(flags & ~REBALANCED_FLAG_MASK);
}
- /** {@inheritDoc} */
- @Override public void selfMarshal() {
+ /** Prepares this partitions map to send to another node. */
+ public void prepareToSend() {
if (!F.isEmpty(parts) && locParts == null)
locParts = copyPartitionsMap(parts);
}
@@ -405,8 +408,8 @@ public class GridDhtPartitionsFullMessage extends
GridDhtPartitionsAbstractMessa
this.topVer = topVer;
}
- /** {@inheritDoc} */
- @Override public void selfUnmarshal() {
+ /** Properly unwraps the partitions map after receiving from another node.
*/
+ public GridDhtPartitionsFullMessage received() {
if (locParts != null && parts == null) {
parts = copyPartitionsMap(locParts);
@@ -438,6 +441,8 @@ public class GridDhtPartitionsFullMessage extends
GridDhtPartitionsAbstractMessa
if (parts == null)
parts = new HashMap<>();
+
+ return this;
}
