Merge branch 'master' into IGNITE-26839

Author: wernerdv

modules/codegen2/src/main/java/org/apache/ignite/internal/MessageSerializerGenerator.java

@@ -183,11 +183,9 @@ private void generateMethods(TypeElement type, List<VariableElement> fields) thr
     /**
      * Generates start of write/read methods:
      * <pre>
-     *     public boolean writeTo(Message m, ByteBuffer buf, MessageWriter writer) {
+     *     public boolean writeTo(Message m, MessageWriter writer) {
      *         TestMessage msg = (TestMessage)m;
      *
-     *         writer.setBuffer(buf);
-     *
      *         if (!writer.isHeaderWritten()) {
      *             if (!writer.writeHeader(msg.directType()))
      *                 return false;
@@ -204,16 +202,13 @@ private void start(TypeElement type, Collection<String> code, boolean write) {
 
         code.add(line(METHOD_JAVADOC));
 
-        code.add(line("@Override public boolean %s(Message m, ByteBuffer buf, %s) {",
+        code.add(line("@Override public boolean %s(Message m, %s) {",
             write ? "writeTo" : "readFrom",
             write ? "MessageWriter writer" : "MessageReader reader"));
 
         indent++;
 
         code.add(line("%s msg = (%s)m;", type.getSimpleName().toString(), type.getSimpleName().toString()));
-        code.add(EMPTY);
-        code.add(line("%s.setBuffer(buf);", write ? "writer" : "reader"));
-
         code.add(EMPTY);
 
         if (write) {
@@ -693,7 +688,6 @@ private void writeClassHeader(Writer writer, String pkgName, String serClsName)
         writer.write(NL);
         writer.write("package " + pkgName + ";" + NL + NL);
 
-        imports.add("java.nio.ByteBuffer");
         imports.add("org.apache.ignite.plugin.extensions.communication.Message");
         imports.add("org.apache.ignite.plugin.extensions.communication.MessageSerializer");
         imports.add("org.apache.ignite.plugin.extensions.communication.MessageWriter");

modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java

@@ -109,7 +109,7 @@
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.DataStreamerUpdatesHandler;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotPartitionsQuickVerifyHandler;
-import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotPartitionsVerifyTaskResult;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotPartitionsVerifyResult;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
 import org.apache.ignite.internal.processors.cache.transactions.TransactionProxyImpl;
@@ -3256,7 +3256,7 @@ public void testCheckSnapshot() throws Exception {
 
         StringBuilder sb = new StringBuilder();
 
-        ((SnapshotPartitionsVerifyTaskResult)h.getLastOperationResult()).print(sb::append);
+        ((SnapshotPartitionsVerifyResult)h.getLastOperationResult()).print(sb::append);
 
         assertContains(log, sb.toString(), "The check procedure has finished, no conflicts have been found");
     }

modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageReader.java

@@ -48,6 +48,9 @@ public class DirectMessageReader implements MessageReader {
     @GridToStringInclude
     private final DirectMessageState<StateItem> state;
 
+    /** Buffer for reading. */
+    private ByteBuffer buf;
+
     /** Whether last field was fully read. */
     private boolean lastRead;
 
@@ -65,9 +68,20 @@ public DirectMessageReader(final MessageFactory msgFactory, IgniteCacheObjectPro
 
     /** {@inheritDoc} */
     @Override public void setBuffer(ByteBuffer buf) {
+        this.buf = buf;
+
         state.item().stream.setBuffer(buf);
     }
 
+    /**
+     * Gets but buffer to read from.
+     *
+     * @return Byte buffer.
+     */
+    public ByteBuffer getBuffer() {
+        return buf;
+    }
+
     /** {@inheritDoc} */
     @Override public byte readByte() {
         DirectByteBufferStream stream = state.item().stream;
@@ -395,6 +409,8 @@ public DirectMessageReader(final MessageFactory msgFactory, IgniteCacheObjectPro
     /** {@inheritDoc} */
     @Override public void beforeInnerMessageRead() {
         state.forward();
+
+        state.item().stream.setBuffer(buf);
     }
 
     /** {@inheritDoc} */

modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java

@@ -47,6 +47,9 @@ public class DirectMessageWriter implements MessageWriter {
     @GridToStringInclude
     private final DirectMessageState<StateItem> state;
 
+    /** Buffer for writing. */
+    private ByteBuffer buf;
+
     /** */
     public DirectMessageWriter(final MessageFactory msgFactory) {
         state = new DirectMessageState<>(StateItem.class, new IgniteOutClosure<StateItem>() {
@@ -58,9 +61,20 @@ public DirectMessageWriter(final MessageFactory msgFactory) {
 
     /** {@inheritDoc} */
     @Override public void setBuffer(ByteBuffer buf) {
+        this.buf = buf;
+
         state.item().stream.setBuffer(buf);
     }
 
+    /**
+     * Gets buffer to write to.
+     *
+     * @return Byte buffer.
+     */
+    public ByteBuffer getBuffer() {
+        return buf;
+    }
+
     /** {@inheritDoc} */
     @Override public boolean writeHeader(short type) {
         DirectByteBufferStream stream = state.item().stream;
@@ -364,6 +378,8 @@ public DirectMessageWriter(final MessageFactory msgFactory) {
     /** {@inheritDoc} */
     @Override public void beforeInnerMessageWrite() {
         state.forward();
+
+        state.item().stream.setBuffer(buf);
     }
 
     /** {@inheritDoc} */

modules/core/src/main/java/org/apache/ignite/internal/direct/stream/DirectByteBufferStream.java

@@ -884,7 +884,7 @@ public void writeMessage(Message msg, MessageWriter writer) {
                 try {
                     writer.beforeInnerMessageWrite();
 
-                    lastFinished = msgFactory.serializer(msg.directType()).writeTo(msg, buf, writer);
+                    lastFinished = msgFactory.serializer(msg.directType()).writeTo(msg, writer);
                 }
                 finally {
                     writer.afterInnerMessageWrite(lastFinished);
@@ -1536,7 +1536,7 @@ public <T extends Message> T readMessage(MessageReader reader) {
             try {
                 reader.beforeInnerMessageRead();
 
-                lastFinished = msgFactory.serializer(msg.directType()).readFrom(msg, buf, reader);
+                lastFinished = msgFactory.serializer(msg.directType()).readFrom(msg, reader);
             }
             finally {
                 reader.afterInnerMessageRead(lastFinished);

modules/core/src/main/java/org/apache/ignite/internal/management/snapshot/SnapshotCheckCommand.java

@@ -18,10 +18,10 @@
 package org.apache.ignite.internal.management.snapshot;
 
 import java.util.function.Consumer;
-import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotPartitionsVerifyTaskResult;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotPartitionsVerifyResult;
 
 /** */
-public class SnapshotCheckCommand extends AbstractSnapshotCommand<SnapshotCheckCommandArg, SnapshotPartitionsVerifyTaskResult> {
+public class SnapshotCheckCommand extends AbstractSnapshotCommand<SnapshotCheckCommandArg, SnapshotPartitionsVerifyResult> {
     /** {@inheritDoc} */
     @Override public String description() {
         return "Check snapshot";
@@ -38,7 +38,7 @@ public class SnapshotCheckCommand extends AbstractSnapshotCommand<SnapshotCheckC
     }
 
     /** {@inheritDoc} */
-    @Override public void printResult(SnapshotCheckCommandArg arg, SnapshotPartitionsVerifyTaskResult res, Consumer<String> printer) {
+    @Override public void printResult(SnapshotCheckCommandArg arg, SnapshotPartitionsVerifyResult res, Consumer<String> printer) {
         res.print(printer);
     }
 }

modules/core/src/main/java/org/apache/ignite/internal/management/snapshot/SnapshotCheckTask.java

@@ -19,7 +19,7 @@
 
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager;
-import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotPartitionsVerifyTaskResult;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotPartitionsVerifyResult;
 import org.apache.ignite.internal.processors.task.GridInternal;
 import org.apache.ignite.internal.util.future.IgniteFutureImpl;
 import org.apache.ignite.internal.visor.VisorJob;
@@ -29,17 +29,17 @@
  * @see IgniteSnapshotManager#checkSnapshot(String, String)
  */
 @GridInternal
-public class SnapshotCheckTask extends VisorOneNodeTask<SnapshotCheckCommandArg, SnapshotPartitionsVerifyTaskResult> {
+public class SnapshotCheckTask extends VisorOneNodeTask<SnapshotCheckCommandArg, SnapshotPartitionsVerifyResult> {
     /** Serial version uid. */
     private static final long serialVersionUID = 0L;
 
     /** {@inheritDoc} */
-    @Override protected VisorJob<SnapshotCheckCommandArg, SnapshotPartitionsVerifyTaskResult> job(SnapshotCheckCommandArg arg) {
+    @Override protected VisorJob<SnapshotCheckCommandArg, SnapshotPartitionsVerifyResult> job(SnapshotCheckCommandArg arg) {
         return new SnapshotCheckJob(arg, debug);
     }
 
     /** */
-    private static class SnapshotCheckJob extends SnapshotJob<SnapshotCheckCommandArg, SnapshotPartitionsVerifyTaskResult> {
+    private static class SnapshotCheckJob extends SnapshotJob<SnapshotCheckCommandArg, SnapshotPartitionsVerifyResult> {
         /** Serial version uid. */
         private static final long serialVersionUID = 0L;
 
@@ -52,7 +52,7 @@ protected SnapshotCheckJob(SnapshotCheckCommandArg arg, boolean debug) {
         }
 
         /** {@inheritDoc} */
-        @Override protected SnapshotPartitionsVerifyTaskResult run(SnapshotCheckCommandArg arg) throws IgniteException {
+        @Override protected SnapshotPartitionsVerifyResult run(SnapshotCheckCommandArg arg) throws IgniteException {
             IgniteSnapshotManager snpMgr = ignite.context().cache().context().snapshotMgr();
 
             return new IgniteFutureImpl<>(snpMgr.checkSnapshot(arg.snapshotName(), arg.src(), arg.increment())).get();

modules/core/src/main/java/org/apache/ignite/internal/managers/communication/ErrorMessage.java

@@ -36,6 +36,7 @@
  * <p>Because raw serialization of throwables is prohibited, you should use this message when it is necessary
  * to transfer some error as part of some message. See {@link MessageProcessor} for details.
  * <p>Currently, under the hood marshalling and unmarshalling is performed by {@link JdkMarshaller}.
+ * <p>If the message serialization fails, wraps this error with own one.
  */
 @SuppressWarnings({"NullableProblems", "unused"})
 public class ErrorMessage implements Message {
@@ -74,8 +75,22 @@ public ErrorMessage(@Nullable Throwable err) {
         try {
             return U.marshal(jdk(), err);
         }
-        catch (IgniteCheckedException e) {
-            throw new IgniteException("Unable to marshal the holding error.", e);
+        catch (IgniteCheckedException e0) {
+            IgniteCheckedException wrappedErr = new IgniteCheckedException(err.getMessage());
+
+            wrappedErr.setStackTrace(err.getStackTrace());
+            wrappedErr.addSuppressed(e0);
+
+            try {
+                return U.marshal(jdk(), wrappedErr);
+            }
+            catch (IgniteCheckedException e1) {
+                IgniteException marshErr = new IgniteException("Unable to marshal the wrapping error.", e1);
+
+                marshErr.addSuppressed(wrappedErr);
+
+                throw marshErr;
+            }
         }
     }
 

modules/core/src/main/java/org/apache/ignite/internal/managers/communication/IgniteMessageFactoryImpl.java

@@ -22,6 +22,8 @@
 import java.util.function.Supplier;
 
 import org.apache.ignite.IgniteException;
+import org.apache.ignite.internal.direct.DirectMessageReader;
+import org.apache.ignite.internal.direct.DirectMessageWriter;
 import org.apache.ignite.plugin.extensions.communication.Message;
 import org.apache.ignite.plugin.extensions.communication.MessageFactory;
 import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
@@ -43,12 +45,22 @@ public class IgniteMessageFactoryImpl implements MessageFactory {
     /** Delegate serialization to {@code Message} methods. */
     private static final MessageSerializer DEFAULT_SERIALIZER = new MessageSerializer() {
         /** {@inheritDoc} */
-        @Override public boolean writeTo(Message msg, ByteBuffer buf, MessageWriter writer) {
+        @Override public boolean writeTo(Message msg, MessageWriter writer) {
+            ByteBuffer buf = null;
+
+            if (writer instanceof DirectMessageWriter)
+                buf = ((DirectMessageWriter)writer).getBuffer();
+
             return msg.writeTo(buf, writer);
         }
 
         /** {@inheritDoc} */
-        @Override public boolean readFrom(Message msg, ByteBuffer buf, MessageReader reader) {
+        @Override public boolean readFrom(Message msg, MessageReader reader) {
+            ByteBuffer buf = null;
+
+            if (reader instanceof DirectMessageReader)
+                buf = ((DirectMessageReader)reader).getBuffer();
+
             return msg.readFrom(buf, reader);
         }
     };

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

@@ -87,7 +87,7 @@ static CacheInvokeDirectResult lazyResult(KeyCacheObject key, Object res) {
      */
     public CacheInvokeDirectResult(KeyCacheObject key, Throwable err) {
         this.key = key;
-        this.errMsg = new ErrorMessage(err);
+        errMsg = new ErrorMessage(err);
     }
 
     /**