diff --git a/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java
index 134e11f53cd50..bc1e5937c304e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java
@@ -67,6 +67,18 @@ public class DirectMessageWriter implements MessageWriter {
     /** Buffer for writing. */
     private ByteBuffer buf;
 
+    /**
+     * Cached stream of the current state item. Avoids re-resolving {@code state.item().stream} on every primitive
+     * write; updated only when the current item changes (buffer set / nested write enter or exit).
+     */
+    private DirectByteBufferStream curStream;
+
+    /** Reusable writer for compressed payload serialization. Lazily created; thread-confined like the writer itself. */
+    private DirectMessageWriter tmpWriter;
+
+    /** Reusable scratch buffer for compressed payload serialization. Retained at the largest size seen so far. */
+    private ByteBuffer tmpBuf;
+
     /** @param msgFactory Message factory. */
     public DirectMessageWriter(final MessageFactory msgFactory) {
         this(msgFactory, DFLT_NETWORK_COMPRESSION);
@@ -91,7 +103,9 @@ public DirectMessageWriter(final MessageFactory msgFactory, final int compressio
     @Override public void setBuffer(ByteBuffer buf) {
         this.buf = buf;
 
-        state.item().stream.setBuffer(buf);
+        curStream = state.item().stream;
+
+        curStream.setBuffer(buf);
     }
 
     /**
@@ -105,7 +119,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeHeader(short type) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeShort(type);
 
@@ -114,7 +128,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeByte(byte val) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeByte(val);
 
@@ -123,7 +137,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeShort(short val) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeShort(val);
 
@@ -132,7 +146,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeInt(int val) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeInt(val);
 
@@ -141,7 +155,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeLong(long val) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeLong(val);
 
@@ -150,7 +164,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeFloat(float val) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeFloat(val);
 
@@ -159,7 +173,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeDouble(double val) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeDouble(val);
 
@@ -168,7 +182,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeChar(char val) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeChar(val);
 
@@ -177,7 +191,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeBoolean(boolean val) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeBoolean(val);
 
@@ -186,7 +200,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeByteArray(@Nullable byte[] val) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeByteArray(val);
 
@@ -195,7 +209,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeByteArray(byte[] val, long off, int len) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeByteArray(val, off, len);
 
@@ -204,7 +218,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeShortArray(@Nullable short[] val) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeShortArray(val);
 
@@ -213,7 +227,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeIntArray(@Nullable int[] val) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeIntArray(val);
 
@@ -222,7 +236,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeLongArray(@Nullable long[] val) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeLongArray(val);
 
@@ -231,7 +245,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeLongArray(long[] val, int len) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeLongArray(val, len);
 
@@ -240,7 +254,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeFloatArray(@Nullable float[] val) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeFloatArray(val);
 
@@ -249,7 +263,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeDoubleArray(@Nullable double[] val) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeDoubleArray(val);
 
@@ -258,7 +272,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeCharArray(@Nullable char[] val) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeCharArray(val);
 
@@ -267,7 +281,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeBooleanArray(@Nullable boolean[] val) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeBooleanArray(val);
 
@@ -276,7 +290,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeString(String val) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeString(val);
 
@@ -285,7 +299,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeBitSet(BitSet val) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeBitSet(val);
 
@@ -294,7 +308,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeUuid(UUID val) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeUuid(val);
 
@@ -303,7 +317,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeIgniteUuid(IgniteUuid val) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeIgniteUuid(val);
 
@@ -312,7 +326,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeAffinityTopologyVersion(AffinityTopologyVersion val) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeAffinityTopologyVersion(val);
 
@@ -321,11 +335,11 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeMessage(@Nullable Message msg, boolean compress) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         if (compress)
             writeCompressedMessage(
-                tmpWriter -> tmpWriter.state.item().stream.writeMessage(msg, tmpWriter),
+                w -> w.state.item().stream.writeMessage(msg, w),
                 msg == null,
                 stream
             );
@@ -337,7 +351,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeCacheObject(@Nullable CacheObject obj) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeCacheObject(obj);
 
@@ -346,7 +360,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeKeyCacheObject(KeyCacheObject obj) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeKeyCacheObject(obj);
 
@@ -355,7 +369,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeGridLongList(@Nullable GridLongList ll) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeGridLongList(ll);
 
@@ -364,7 +378,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public <T> boolean writeObjectArray(T[] arr, MessageArrayType type) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeObjectArray(arr, type, this);
 
@@ -373,7 +387,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public <T> boolean writeCollection(Collection<T> col, MessageCollectionType type) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeCollection(col, type, this);
 
@@ -382,11 +396,11 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public <K, V> boolean writeMap(Map<K, V> map, MessageMapType type, boolean compress) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         if (compress)
             writeCompressedMessage(
-                tmpWriter -> tmpWriter.state.item().stream.writeMap(map, type, tmpWriter),
+                w -> w.state.item().stream.writeMap(map, type, w),
                 map == null,
                 stream
             );
@@ -398,7 +412,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeIgniteProductVersion(IgniteProductVersion ver) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeIgniteProductVersion(ver);
 
@@ -407,7 +421,7 @@ public ByteBuffer getBuffer() {
 
     /** {@inheritDoc} */
     @Override public boolean writeGridCacheVersion(GridCacheVersion ver) {
-        DirectByteBufferStream stream = state.item().stream;
+        DirectByteBufferStream stream = curStream;
 
         stream.writeGridCacheVersion(ver);
 
@@ -443,12 +457,16 @@ public ByteBuffer getBuffer() {
     @Override public void beforeNestedWrite() {
         state.forward();
 
-        state.item().stream.setBuffer(buf);
+        curStream = state.item().stream;
+
+        curStream.setBuffer(buf);
     }
 
     /** {@inheritDoc} */
     @Override public void afterNestedWrite(boolean finished) {
         state.backward(finished);
+
+        curStream = state.item().stream;
     }
 
     /** {@inheritDoc} */
@@ -474,9 +492,19 @@ private void writeCompressedMessage(Consumer<DirectMessageWriter> consumer, bool
         }
 
         if (!stream.serializeFinished()) {
-            ByteBuffer tmpBuf = ByteBuffer.allocateDirect(TMP_BUF_CAPACITY);
-
-            DirectMessageWriter tmpWriter = new DirectMessageWriter(msgFactory, compressionLvl);
+            // CompressedMessage consumes the scratch buffer in its constructor (deflates into its own byte array),
+            // so the buffer never escapes this method and can be reused across fields; heap is cheaper than direct
+            // here (no native alloc / Cleaner churn).
+            if (tmpBuf == null)
+                tmpBuf = ByteBuffer.allocate(TMP_BUF_CAPACITY);
+            else
+                tmpBuf.clear();
+
+            // Reuse the temp writer across fields/messages instead of allocating a fresh state stack each time.
+            if (tmpWriter == null)
+                tmpWriter = new DirectMessageWriter(msgFactory, compressionLvl);
+            else
+                tmpWriter.reset();
 
             tmpWriter.setBuffer(tmpBuf);
 
@@ -484,14 +512,12 @@ private void writeCompressedMessage(Consumer<DirectMessageWriter> consumer, bool
 
             do {
                 if (tmpBuf.remaining() <= tmpBuf.capacity() / 10) {
-                    byte[] bytes = new byte[tmpBuf.position()];
+                    ByteBuffer newBuf = ByteBuffer.allocate(tmpBuf.capacity() * 2);
 
                     tmpBuf.flip();
-                    tmpBuf.get(bytes);
-
-                    tmpBuf = ByteBuffer.allocateDirect(tmpBuf.capacity() * 2);
+                    newBuf.put(tmpBuf);
 
-                    tmpBuf.put(bytes);
+                    tmpBuf = newBuf;
 
                     tmpWriter.setBuffer(tmpBuf);
                 }