IGNITE-28985 GridNearTxPrepareRequest might not implement undesirable SelfMarshallingMessage

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxPrepareRequest.java

@@ -91,12 +91,12 @@ public class GridDistributedTxPrepareRequest extends GridDistributedBaseMessage
     /** Transaction read set. */
     @Order(5)
     @GridToStringInclude
-    public Collection<IgniteTxEntry> reads;
+    public @Nullable Collection<IgniteTxEntry> reads;
 
     /** Transaction write entries. */
     @Order(6)
     @GridToStringInclude
-    public Collection<IgniteTxEntry> writes;
+    public @Nullable Collection<IgniteTxEntry> writes;
 
     /** Keys whose DHT version has to be verified on the remote node. */
     @Order(7)
@@ -148,7 +148,7 @@ public GridDistributedTxPrepareRequest(
         IgniteInternalTx tx,
         long timeout,
         @Nullable Collection<IgniteTxEntry> reads,
-        Collection<IgniteTxEntry> writes,
+        @Nullable Collection<IgniteTxEntry> writes,
         Map<UUID, Collection<UUID>> txNodes,
         boolean retVal,
         boolean last,
@@ -289,14 +289,14 @@ public TransactionIsolation isolation() {
     /**
      * @return Read set.
      */
-    public Collection<IgniteTxEntry> reads() {
+    public @Nullable Collection<IgniteTxEntry> reads() {
         return reads;
     }
 
     /**
      * @return Write entries.
      */
-    public Collection<IgniteTxEntry> writes() {
+    public @Nullable Collection<IgniteTxEntry> writes() {
         return writes;
     }
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java

@@ -146,15 +146,9 @@ public GridDhtTxPrepareRequest(
         int taskNameHash,
         boolean storeWriteThrough,
         boolean retVal,
-        Collection<PartitionUpdateCountersMessage> updCntrs) {
-        super(tx,
-            timeout,
-            null,
-            dhtWrites,
-            txNodes,
-            retVal,
-            last,
-            onePhaseCommit);
+        Collection<PartitionUpdateCountersMessage> updCntrs
+    ) {
+        super(tx, timeout, null, dhtWrites, txNodes, retVal, last, onePhaseCommit);
 
         assert futId != null;
         assert miniId != 0;

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java

@@ -510,27 +510,21 @@ private GridNearTxPrepareRequest createRequest(
         MiniFuture fut,
         long timeout,
         Collection<IgniteTxEntry> reads,
-        Collection<IgniteTxEntry> writes) {
+        Collection<IgniteTxEntry> writes
+    ) {
         GridDistributedTxMapping m = fut.mapping();
 
-        GridNearTxPrepareRequest req = new GridNearTxPrepareRequest(
-            futId,
-            tx.topologyVersion(),
-            tx,
-            timeout,
+        GridNearTxPrepareRequest req = createPrepareRequest(
+            txNodes,
+            m,
             reads,
             writes,
-            m.hasNearCacheEntries(),
-            txNodes,
+            timeout,
             m.last(),
             tx.onePhaseCommit(),
-            tx.needReturnValue() && tx.implicit(),
-            tx.implicitSingle(),
-            m.explicitLock(),
-            tx.taskNameHash(),
             m.clientFirst(),
-            txNodes.size() == 1,
-            tx.txState().recovery());
+            txNodes.size() == 1
+        );
 
         for (IgniteTxEntry txEntry : writes) {
             if (txEntry.op() == TRANSFORM)

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java

@@ -507,24 +507,17 @@ private void proceedPrepare(GridDistributedTxMapping m, @Nullable final Queue<Gr
             long timeout = tx.remainingTime();
 
             if (timeout != -1) {
-                GridNearTxPrepareRequest req = new GridNearTxPrepareRequest(
-                    futId,
-                    tx.topologyVersion(),
-                    tx,
-                    timeout,
+                GridNearTxPrepareRequest req = createPrepareRequest(
+                    txMapping.transactionNodes(),
+                    m,
                     null,
                     m.writes(),
-                    m.hasNearCacheEntries(),
-                    txMapping.transactionNodes(),
+                    timeout,
                     m.last(),
                     tx.onePhaseCommit(),
-                    tx.needReturnValue() && tx.implicit(),
-                    tx.implicitSingle(),
-                    m.explicitLock(),
-                    tx.taskNameHash(),
                     m.clientFirst(),
-                    txMapping.transactionNodes().size() == 1,
-                    tx.txState().recovery());
+                    txMapping.transactionNodes().size() == 1
+                );
 
                 for (IgniteTxEntry txEntry : m.entries()) {
                     if (txEntry.op() == TRANSFORM)
@@ -556,7 +549,7 @@ private void proceedPrepare(GridDistributedTxMapping m, @Nullable final Queue<Gr
                         m.hasNearCacheEntries() ? cctx.tm().txHandler().prepareNearTxLocal(tx, req)
                         : cctx.tm().txHandler().prepareColocatedTx(tx, req);
 
-                    prepFut.listen(new CI1<IgniteInternalFuture<GridNearTxPrepareResponse>>() {
+                    prepFut.listen(new CI1<>() {
                         @Override public void apply(IgniteInternalFuture<GridNearTxPrepareResponse> prepFut) {
                             try {
                                 fut.onResult(prepFut.get());

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java

@@ -205,29 +205,24 @@ private MiniFuture miniFuture(int miniId) {
      * @param writes Writes.
      * @return Request.
      */
-    private GridNearTxPrepareRequest createRequest(Map<UUID, Collection<UUID>> txNodes,
+    private GridNearTxPrepareRequest createRequest(
+        Map<UUID, Collection<UUID>> txNodes,
         GridDistributedTxMapping m,
         long timeout,
         Collection<IgniteTxEntry> reads,
-        Collection<IgniteTxEntry> writes) {
-        GridNearTxPrepareRequest req = new GridNearTxPrepareRequest(
-            futId,
-            tx.topologyVersion(),
-            tx,
-            timeout,
+        Collection<IgniteTxEntry> writes
+    ) {
+        GridNearTxPrepareRequest req = createPrepareRequest(
+            txNodes,
+            m,
             reads,
             writes,
-            m.hasNearCacheEntries(),
-            txNodes,
+            timeout,
             true,
             tx.onePhaseCommit(),
-            tx.needReturnValue() && tx.implicit(),
-            tx.implicitSingle(),
-            m.explicitLock(),
-            tx.taskNameHash(),
             false,
-            true,
-            tx.txState().recovery());
+            true
+        );
 
         for (IgniteTxEntry txEntry : writes) {
             if (txEntry.op() == TRANSFORM)
@@ -258,7 +253,7 @@ private void prepareLocal(GridNearTxPrepareRequest req,
             cctx.tm().txHandler().prepareNearTxLocal(tx, req) :
             cctx.tm().txHandler().prepareColocatedTx(tx, req);
 
-        prepFut.listen(new CI1<IgniteInternalFuture<GridNearTxPrepareResponse>>() {
+        prepFut.listen(new CI1<>() {
             @Override public void apply(IgniteInternalFuture<GridNearTxPrepareResponse> prepFut) {
                 try {
                     fut.onResult(prepFut.get(), nearEntries);

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFutureAdapter.java

@@ -43,6 +43,8 @@
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteReducer;
 import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.jetbrains.annotations.Nullable;
 
 import static org.apache.ignite.internal.processors.cache.GridCacheOperation.NOOP;
 
@@ -191,6 +193,63 @@ final void checkOnePhase(Map<UUID, Collection<UUID>> txNodes) {
         }
     }
 
+    /**
+     * Creates {@link GridNearTxPrepareRequest} and prepares as {@link Message} to send to another node. Affects
+     * {@code writes} and {@code reads} (if aren't empty).
+     *
+     * @param txNodes Transaction nodes mapping.
+     * @param mapping Distributed transaction mapping.
+     * @param reads Read entries.
+     * @param writes Write entries.
+     * @param timeout Transaction timeout.
+     * @param last {@code True} if this last prepare request for node.
+     * @param onePhaseCommit One phase commit flag.
+     * @param firstClientReq {@code True} if first optimistic tx prepare request sent from client node.
+     * @param allowWaitTopFut {@code True} if it is safe for first client request to wait for topology future.
+     */
+    protected GridNearTxPrepareRequest createPrepareRequest(
+        Map<UUID, Collection<UUID>> txNodes,
+        GridDistributedTxMapping mapping,
+        @Nullable Collection<IgniteTxEntry> reads,
+        @Nullable Collection<IgniteTxEntry> writes,
+        long timeout,
+        boolean last,
+        boolean onePhaseCommit,
+        boolean firstClientReq,
+        boolean allowWaitTopFut
+    ) {
+        // Of all tx messages, only the near prepare request transfers entry expiry policies.
+        if (!F.isEmpty(writes)) {
+            for (IgniteTxEntry we : writes)
+                we.transferExpiryPolicy(true);
+        }
+
+        if (!F.isEmpty(reads)) {
+            for (IgniteTxEntry re : reads)
+                re.transferExpiryPolicy(true);
+        }
+
+        return new GridNearTxPrepareRequest(
+            futId,
+            tx.topologyVersion(),
+            tx,
+            timeout,
+            reads,
+            writes,
+            mapping.hasNearCacheEntries(),
+            txNodes,
+            last,
+            onePhaseCommit,
+            tx.needReturnValue() && tx.implicit(),
+            tx.implicitSingle(),
+            mapping.explicitLock(),
+            tx.taskNameHash(),
+            firstClientReq,
+            allowWaitTopFut,
+            tx.txState().recovery()
+        );
+    }
+
     /**
      * @param m Mapping.
      * @param res Response.

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareRequest.java

@@ -23,7 +23,6 @@
 import java.util.Map;
 import java.util.UUID;
 import org.apache.ignite.internal.Order;
-import org.apache.ignite.internal.SelfMarshallingMessage;
 import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
 import org.apache.ignite.internal.processors.cache.GridCacheContext;
 import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxPrepareRequest;
@@ -39,7 +38,7 @@
 /**
  * Near transaction prepare request to primary node. 'Near' means 'Initiating node' here, not 'Near Cache'.
  */
-public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest implements SelfMarshallingMessage {
+public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
     /** */
     private static final int NEAR_FLAG_MASK = 0x01;
 
@@ -114,8 +113,8 @@ public GridNearTxPrepareRequest(
         AffinityTopologyVersion topVer,
         GridNearTxLocal tx,
         long timeout,
-        Collection<IgniteTxEntry> reads,
-        Collection<IgniteTxEntry> writes,
+        @Nullable Collection<IgniteTxEntry> reads,
+        @Nullable Collection<IgniteTxEntry> writes,
         boolean near,
         Map<UUID, Collection<UUID>> txNodes,
         boolean last,
@@ -128,14 +127,7 @@ public GridNearTxPrepareRequest(
         boolean allowWaitTopFut,
         boolean recovery
     ) {
-        super(tx,
-            timeout,
-            reads,
-            writes,
-            txNodes,
-            retVal,
-            last,
-            onePhaseCommit);
+        super(tx, timeout, reads, writes, txNodes, retVal, last, onePhaseCommit);
 
         assert futId != null;
         assert !firstClientReq || tx.optimistic() : tx;
@@ -292,25 +284,6 @@ private boolean isFlag(int mask) {
         return U.safeAbs(version().hashCode());
     }
 
-    /** {@inheritDoc} */
-    @Override public void selfMarshal() {
-        // Of all tx messages, only the near prepare request transfers entry expiry policies.
-        if (writes() != null) {
-            for (IgniteTxEntry e : writes())
-                e.transferExpiryPolicy(true);
-        }
-
-        if (reads() != null) {
-            for (IgniteTxEntry e : reads())
-                e.transferExpiryPolicy(true);
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override public void selfUnmarshal() {
-        // No-op.
-    }
-
     /** {@inheritDoc} */
     @Override public String toString() {
         StringBuilder flags = new StringBuilder();