You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by na...@apache.org on 2022/01/19 17:23:10 UTC
[ignite] branch master updated: IGNITE-16327 Fixed NPE in the distributed process on message unmarshal. (#9746)
This is an automated email from the ASF dual-hosted git repository.
namelchev 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 65f202c IGNITE-16327 Fixed NPE in the distributed process on message unmarshal. (#9746)
65f202c is described below
commit 65f202c96f1bf2ef50b0b360e105bc8f28074322
Author: Pavel Pereslegin <xx...@gmail.com>
AuthorDate: Wed Jan 19 20:22:30 2022 +0300
IGNITE-16327 Fixed NPE in the distributed process on message unmarshal. (#9746)
---
.../util/distributed/SingleNodeMessage.java | 8 +-
.../direct/DirectMarshallingMessagesTest.java | 109 +++++++++++++++++++++
.../testsuites/IgniteMarshallerSelfTestSuite.java | 4 +-
3 files changed, 118 insertions(+), 3 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/SingleNodeMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/SingleNodeMessage.java
index dab288e..6d46e60 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/SingleNodeMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/SingleNodeMessage.java
@@ -135,19 +135,23 @@ public class SingleNodeMessage<R extends Serializable> implements Message {
reader.incrementState();
case 2:
- resp = U.fromBytes(reader.readByteArray("data"));
+ byte[] dataBytes = reader.readByteArray("data");
if (!reader.isLastRead())
return false;
+ resp = U.fromBytes(dataBytes);
+
reader.incrementState();
case 3:
- err = U.fromBytes(reader.readByteArray("err"));
+ byte[] errBytes = reader.readByteArray("err");
if (!reader.isLastRead())
return false;
+ err = U.fromBytes(errBytes);
+
reader.incrementState();
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/direct/DirectMarshallingMessagesTest.java b/modules/core/src/test/java/org/apache/ignite/internal/direct/DirectMarshallingMessagesTest.java
new file mode 100644
index 0000000..698f0ff
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/direct/DirectMarshallingMessagesTest.java
@@ -0,0 +1,109 @@
+/*
+ * 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.direct;
+
+import java.nio.ByteBuffer;
+import java.util.UUID;
+import java.util.function.Function;
+import org.apache.ignite.internal.managers.communication.GridIoMessageFactory;
+import org.apache.ignite.internal.managers.communication.IgniteMessageFactoryImpl;
+import org.apache.ignite.internal.util.distributed.SingleNodeMessage;
+import org.apache.ignite.plugin.extensions.communication.IgniteMessageFactory;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.Test;
+
+import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.TEST_PROCESS;
+
+/**
+ * Messages marshalling test.
+ */
+public class DirectMarshallingMessagesTest extends GridCommonAbstractTest {
+ /** Protocol version. */
+ private static final byte PROTO_VER = 2;
+
+ /** Message factory. */
+ private final IgniteMessageFactory msgFactory =
+ new IgniteMessageFactoryImpl(new MessageFactory[] {new GridIoMessageFactory()});
+
+ /** */
+ @Test
+ public void testSingleNodeMessage() {
+ SingleNodeMessage<?> srcMsg =
+ new SingleNodeMessage<>(UUID.randomUUID(), TEST_PROCESS, "data", new Exception("error"));
+
+ SingleNodeMessage<?> resMsg = doMarshalUnmarshal(srcMsg);
+
+ assertEquals(srcMsg.type(), resMsg.type());
+ assertEquals(srcMsg.processId(), resMsg.processId());
+ assertEquals(srcMsg.response(), resMsg.response());
+ assertEquals(srcMsg.error().getClass(), resMsg.error().getClass());
+ assertEquals(srcMsg.error().getMessage(), resMsg.error().getMessage());
+ }
+
+ /**
+ * @param srcMsg Message to marshal.
+ * @param <T> Message type.
+ * @return Unmarshalled message.
+ */
+ private <T extends Message> T doMarshalUnmarshal(T srcMsg) {
+ ByteBuffer buf = ByteBuffer.allocate(8 * 1024);
+
+ boolean fullyWritten = loopBuffer(buf, 0, buf0 -> srcMsg.writeTo(buf0, new DirectMessageWriter(PROTO_VER)));
+ assertTrue("The message was not written completely.", fullyWritten);
+
+ buf.flip();
+
+ byte b0 = buf.get();
+ byte b1 = buf.get();
+
+ short type = (short)((b1 & 0xFF) << 8 | b0 & 0xFF);
+
+ assertEquals(srcMsg.directType(), type);
+
+ T resMsg = (T)msgFactory.create(type);
+
+ boolean fullyRead = loopBuffer(buf, buf.position(),
+ buf0 -> resMsg.readFrom(buf0, new DirectMessageReader(msgFactory, PROTO_VER)));
+ assertTrue("The message was not read completely.", fullyRead);
+
+ return resMsg;
+ }
+
+ /**
+ * @param buf Byte buffer.
+ * @param start Start position.
+ * @param func Function that is sequentially executed on a different-sized part of the buffer.
+ * @return {@code True} if the function returns {@code True} at least once, {@code False} otherwise.
+ */
+ private boolean loopBuffer(ByteBuffer buf, int start, Function<ByteBuffer, Boolean> func) {
+ int pos = start;
+
+ do {
+ buf.position(start);
+ buf.limit(++pos);
+
+ if (func.apply(buf))
+ return true;
+ }
+ while (pos < buf.capacity());
+
+ return false;
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteMarshallerSelfTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteMarshallerSelfTestSuite.java
index 8e22bfb..48040d6 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteMarshallerSelfTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteMarshallerSelfTestSuite.java
@@ -17,6 +17,7 @@
package org.apache.ignite.testsuites;
+import org.apache.ignite.internal.direct.DirectMarshallingMessagesTest;
import org.apache.ignite.internal.direct.stream.v2.DirectByteBufferStreamImplV2ByteOrderSelfTest;
import org.apache.ignite.internal.marshaller.optimized.OptimizedMarshallerEnumSelfTest;
import org.apache.ignite.internal.marshaller.optimized.OptimizedMarshallerNodeFailoverTest;
@@ -50,7 +51,8 @@ import org.junit.runners.Suite;
DirectByteBufferStreamImplV2ByteOrderSelfTest.class,
GridHandleTableSelfTest.class,
OptimizedMarshallerPooledSelfTest.class,
- MarshallerEnumDeadlockMultiJvmTest.class
+ MarshallerEnumDeadlockMultiJvmTest.class,
+ DirectMarshallingMessagesTest.class
})
public class IgniteMarshallerSelfTestSuite {
}