anton-vinogradov commented on code in PR #13095:
URL: https://github.com/apache/ignite/pull/13095#discussion_r3658464393
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java:
##########
@@ -1184,7 +1198,29 @@ private void processDhtTxPrepareRequest(final UUID
nodeId, final GridDhtTxPrepar
if (nearTx != null)
res.nearEvicted(nearTx.evicted());
- List<IgniteTxKey> writesCacheMissed = req.nearWritesCacheMissed();
+ List<IgniteTxKey> writesCacheMissed = new ArrayList<>();
Review Comment:
Two things there were mine and both are now gone. `e.initializeContext(ctx,
req.topologyVersion(), true)` was dead: the line above it does
`e.context(cacheCtx)`, and initializeContext starts with `if (this.ctx ==
null)`, so it was a guaranteed no-op — the real initialization of near entries
happens in GridNearTxRemote.addEntries, which also handles the destroyed-cache
case. And `writesCacheMissed != null` was always true here, since the list is
now built locally instead of coming from req.nearWritesCacheMissed() (which
master allowed to be null): that made an empty nearEvicted collection travel
back on every prepare response. It is `!writesCacheMissed.isEmpty()` now, i.e.
master's behaviour.
The loop itself moved here from GridDhtTxPrepareRequest#finishUnmarshal,
where master bound the entry contexts during unmarshalling; with codegen
serialization the binding belongs to the handler. The extra
topologyVersion().before(startTopologyVersion()) condition is the recreate fix
covered by GridNearTxRecreateEvictionTest.
##########
modules/core/src/test/java/org/apache/ignite/plugin/extensions/communication/MessageMarshalOnceTest.java:
##########
@@ -0,0 +1,256 @@
+/*
+ * 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.plugin.extensions.communication;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.CoreMessagesProvider;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.GridTopic;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.managers.communication.GridIoPolicy;
+import org.apache.ignite.internal.processors.cache.CacheObjectContext;
+import org.apache.ignite.internal.processors.cache.GridCacheMessage;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.plugin.AbstractTestPluginProvider;
+import org.apache.ignite.plugin.ExtensionRegistry;
+import org.apache.ignite.plugin.PluginContext;
+import org.apache.ignite.spi.IgniteSpiException;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.Test;
+
+/**
+ * Verifies the marshal-once-before-fan-out contract of the collection {@code
send}: a message broadcast to N remote
+ * nodes is marshalled <b>exactly once</b> (prepared before the fan-out and
reused for every destination), not once per
+ * destination. A counting marshaller is registered for a test message;
broadcasting it to {@link #RMT_CNT} remote nodes
+ * must produce {@code RMT_CNT} sends but a single {@code marshal}. Counting
the sends keeps the check honest (a
+ * lone send would also marshal once). The unmarshal-once counterpart lives in
{@link MessageUnmarshalOnceTest}.
+ *
+ * <p>The second scenario covers the send-retry path: a cache message whose
first send attempt fails is retried,
+ * and the retry must not re-marshal the payload (it is prepared once by
{@code GridCacheIoManager} before the
+ * retry loop, while {@code GridIoManager} marshals only the wrapper).
+ */
+public class MessageMarshalOnceTest extends GridCommonAbstractTest {
Review Comment:
They test different things with different lifecycles, so I'd rather keep
them apart. MessageMarshalOnceTest verifies a production contract of the send
path — sendToMany/prepare/sendWithRetry marshal a message once, not per
destination and not per retry — and for that it starts real 5- and 2-node grids
with a plugin registering custom direct types and an SPI that fails the first
send. The other one verifies the test-only detector itself and that the
suite-wide flag is on; it starts no grid and needs no configuration at all, so
merging would put those unit checks under a configuration with a plugin and a
custom communication SPI.
The javadoc calling it "the unmarshal-once counterpart" was misleading —
that is what makes them look like halves of one test. Reworded, and renamed the
class to MessageUnmarshalOnceCheckTest so the subject is explicit.
##########
modules/core/src/test/java/org/apache/ignite/plugin/extensions/communication/MessageMarshalOnceTest.java:
##########
@@ -0,0 +1,256 @@
+/*
+ * 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.plugin.extensions.communication;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.CoreMessagesProvider;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.GridTopic;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.managers.communication.GridIoPolicy;
+import org.apache.ignite.internal.processors.cache.CacheObjectContext;
+import org.apache.ignite.internal.processors.cache.GridCacheMessage;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.plugin.AbstractTestPluginProvider;
+import org.apache.ignite.plugin.ExtensionRegistry;
+import org.apache.ignite.plugin.PluginContext;
+import org.apache.ignite.spi.IgniteSpiException;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.Test;
+
+/**
+ * Verifies the marshal-once-before-fan-out contract of the collection {@code
send}: a message broadcast to N remote
+ * nodes is marshalled <b>exactly once</b> (prepared before the fan-out and
reused for every destination), not once per
+ * destination. A counting marshaller is registered for a test message;
broadcasting it to {@link #RMT_CNT} remote nodes
+ * must produce {@code RMT_CNT} sends but a single {@code marshal}. Counting
the sends keeps the check honest (a
+ * lone send would also marshal once). The unmarshal-once counterpart lives in
{@link MessageUnmarshalOnceTest}.
+ *
+ * <p>The second scenario covers the send-retry path: a cache message whose
first send attempt fails is retried,
+ * and the retry must not re-marshal the payload (it is prepared once by
{@code GridCacheIoManager} before the
+ * retry loop, while {@code GridIoManager} marshals only the wrapper).
+ */
+public class MessageMarshalOnceTest extends GridCommonAbstractTest {
+ /** Direct type for the test message, past the core range. */
+ private static final short TYPE =
(short)(CoreMessagesProvider.MAX_MESSAGE_ID + 1);
+
+ /** Direct type for the retry-check cache message. */
+ private static final short RETRY_TYPE =
(short)(CoreMessagesProvider.MAX_MESSAGE_ID + 2);
+
+ /** Number of remote destinations to broadcast to (a single marshal must
serve all of them). */
+ private static final int RMT_CNT = 4;
+
+ /** Counts {@code marshal} invocations of {@link MarshalOnceCheckMessage}
across the JVM. */
+ private static final AtomicInteger MARSHAL_CNT = new AtomicInteger();
+
+ /** Counts {@code marshal} invocations of {@link RetryCheckMessage} across
the JVM. */
+ private static final AtomicInteger RETRY_MARSHAL_CNT = new AtomicInteger();
+
+ /** When {@code true}, nodes are configured with the SPI failing the first
send of {@link RetryCheckMessage}. */
+ private boolean failFirstSend;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.setCommunicationSpi(failFirstSend ? new FailFirstSendSpi() : new
TestRecordingCommunicationSpi());
+
+ cfg.setPluginProviders(new AbstractTestPluginProvider() {
+ @Override public String name() {
+ return "marshal-once-test";
+ }
+
+ @Override public void initExtensions(PluginContext ctx,
ExtensionRegistry registry) {
+ registry.registerExtension(MessageFactoryProvider.class,
factory -> {
+ factory.register(TYPE, MarshalOnceCheckMessage::new, new
Serializer(), new CountingMarshaller());
+ factory.register(RETRY_TYPE, RetryCheckMessage::new, new
RetrySerializer(), new RetryCountingMarshaller());
+ });
+ }
+ });
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+
+ super.afterTest();
+ }
+
+ /** @throws Exception If failed. */
+ @Test
+ public void testBroadcastMarshalsExactlyOnce() throws Exception {
+ startGrids(RMT_CNT + 1);
+
+ TestRecordingCommunicationSpi commSpi =
TestRecordingCommunicationSpi.spi(grid(0));
+
+ commSpi.record(MarshalOnceCheckMessage.class);
+
+ List<ClusterNode> rmts = new ArrayList<>(RMT_CNT);
+
+ for (int i = 1; i <= RMT_CNT; i++)
+ rmts.add(grid(i).localNode());
+
+ MARSHAL_CNT.set(0);
Review Comment:
In a clean run it cannot: MARSHAL_CNT is incremented only by
CountingMarshaller, which is registered for direct type MAX_MESSAGE_ID + 1, and
that type belongs only to MarshalOnceCheckMessage — created nowhere but in this
test method. The reset matters because the counter is static, i.e. shared by
all five nodes in the JVM and surviving across method boundaries: with @Repeat
(GridAbstractTest installs RepeatRule) or a repeated run of the class in the
same fork the second iteration would enter with a non-zero value and the assert
below would fail for the wrong reason.
##########
modules/core/src/test/java/org/apache/ignite/plugin/extensions/communication/MessageMarshalOnceTest.java:
##########
@@ -0,0 +1,256 @@
+/*
+ * 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.plugin.extensions.communication;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.CoreMessagesProvider;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.GridTopic;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.managers.communication.GridIoPolicy;
+import org.apache.ignite.internal.processors.cache.CacheObjectContext;
+import org.apache.ignite.internal.processors.cache.GridCacheMessage;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.plugin.AbstractTestPluginProvider;
+import org.apache.ignite.plugin.ExtensionRegistry;
+import org.apache.ignite.plugin.PluginContext;
+import org.apache.ignite.spi.IgniteSpiException;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.Test;
+
+/**
+ * Verifies the marshal-once-before-fan-out contract of the collection {@code
send}: a message broadcast to N remote
+ * nodes is marshalled <b>exactly once</b> (prepared before the fan-out and
reused for every destination), not once per
+ * destination. A counting marshaller is registered for a test message;
broadcasting it to {@link #RMT_CNT} remote nodes
+ * must produce {@code RMT_CNT} sends but a single {@code marshal}. Counting
the sends keeps the check honest (a
+ * lone send would also marshal once). The unmarshal-once counterpart lives in
{@link MessageUnmarshalOnceTest}.
+ *
+ * <p>The second scenario covers the send-retry path: a cache message whose
first send attempt fails is retried,
+ * and the retry must not re-marshal the payload (it is prepared once by
{@code GridCacheIoManager} before the
+ * retry loop, while {@code GridIoManager} marshals only the wrapper).
+ */
+public class MessageMarshalOnceTest extends GridCommonAbstractTest {
+ /** Direct type for the test message, past the core range. */
+ private static final short TYPE =
(short)(CoreMessagesProvider.MAX_MESSAGE_ID + 1);
+
+ /** Direct type for the retry-check cache message. */
+ private static final short RETRY_TYPE =
(short)(CoreMessagesProvider.MAX_MESSAGE_ID + 2);
+
+ /** Number of remote destinations to broadcast to (a single marshal must
serve all of them). */
+ private static final int RMT_CNT = 4;
+
+ /** Counts {@code marshal} invocations of {@link MarshalOnceCheckMessage}
across the JVM. */
+ private static final AtomicInteger MARSHAL_CNT = new AtomicInteger();
+
+ /** Counts {@code marshal} invocations of {@link RetryCheckMessage} across
the JVM. */
+ private static final AtomicInteger RETRY_MARSHAL_CNT = new AtomicInteger();
+
+ /** When {@code true}, nodes are configured with the SPI failing the first
send of {@link RetryCheckMessage}. */
+ private boolean failFirstSend;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.setCommunicationSpi(failFirstSend ? new FailFirstSendSpi() : new
TestRecordingCommunicationSpi());
+
+ cfg.setPluginProviders(new AbstractTestPluginProvider() {
+ @Override public String name() {
+ return "marshal-once-test";
+ }
+
+ @Override public void initExtensions(PluginContext ctx,
ExtensionRegistry registry) {
+ registry.registerExtension(MessageFactoryProvider.class,
factory -> {
+ factory.register(TYPE, MarshalOnceCheckMessage::new, new
Serializer(), new CountingMarshaller());
+ factory.register(RETRY_TYPE, RetryCheckMessage::new, new
RetrySerializer(), new RetryCountingMarshaller());
+ });
+ }
+ });
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+
+ super.afterTest();
+ }
+
+ /** @throws Exception If failed. */
+ @Test
+ public void testBroadcastMarshalsExactlyOnce() throws Exception {
+ startGrids(RMT_CNT + 1);
+
+ TestRecordingCommunicationSpi commSpi =
TestRecordingCommunicationSpi.spi(grid(0));
+
+ commSpi.record(MarshalOnceCheckMessage.class);
+
+ List<ClusterNode> rmts = new ArrayList<>(RMT_CNT);
+
+ for (int i = 1; i <= RMT_CNT; i++)
+ rmts.add(grid(i).localNode());
+
+ MARSHAL_CNT.set(0);
+
+ grid(0).context().io().sendToGridTopic(rmts, GridTopic.TOPIC_IO_TEST,
new MarshalOnceCheckMessage(),
+ GridIoPolicy.PUBLIC_POOL);
+
+ assertEquals("Broadcast must be sent to all " + RMT_CNT + " nodes",
RMT_CNT, commSpi.recordedMessages(true).size());
+
+ assertEquals("A message broadcast to " + RMT_CNT + " nodes must be
marshalled exactly once, not per destination",
+ 1, MARSHAL_CNT.get());
+ }
+
+ /** @throws Exception If failed. */
+ @Test
+ public void testRetryMarshalsExactlyOnce() throws Exception {
+ failFirstSend = true;
+
+ startGrids(2);
+
+ RETRY_MARSHAL_CNT.set(0);
Review Comment:
Same as the other one: RETRY_MARSHAL_CNT is touched only by
RetryCountingMarshaller, registered for direct type MAX_MESSAGE_ID + 2, which
belongs only to RetryCheckMessage created in this method — so it is 0 in a
clean run. The reset guards against the static counter carrying a value over
from a repeated run in the same JVM.
##########
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/message/CalciteMessageUnmarshalThreadIntegrationTest.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * 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.processors.query.calcite.message;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import
org.apache.ignite.internal.processors.query.calcite.exec.task.AbstractQueryTaskExecutor;
+import
org.apache.ignite.internal.processors.query.calcite.integration.AbstractBasicIntegrationTest;
+import org.junit.Test;
+
+/**
+ * Verifies that payload of calcite messages is unmarshalled on the query task
executor, not on a NIO thread: probe
+ * objects travel as {@code QueryStartRequest} parameters and as {@code
QueryBatchMessage} rows, and record the thread
+ * that deserializes them.
+ */
+public class CalciteMessageUnmarshalThreadIntegrationTest extends
AbstractBasicIntegrationTest {
Review Comment:
They can't share a class — the bases are incompatible.
CalciteCommunicationMessageSerializationTest extends
AbstractMessageSerializationTest, a plain JUnit class that starts no nodes: it
walks the registered direct types and round-trips each message through mock
writer/reader, and it runs in the unit-level suite. The unmarshal-thread test
extends AbstractBasicIntegrationTest (GridCommonAbstractTest), starts two
servers plus a client and needs real SQL — CREATE TABLE with an OTHER column,
200 inserts and a query with a custom-object parameter — because it asserts on
the threads in which the payload is actually deserialized. Putting it there
would mean dragging the grid lifecycle into a unit-level serialization test and
moving that class out of its suite.
##########
modules/codegen/src/main/java/org/apache/ignite/internal/MessageProcessor.java:
##########
@@ -97,21 +110,32 @@ public class MessageProcessor extends AbstractProcessor {
static final String[] SKIP_MESSAGES = {
Review Comment:
A Class-typed API cannot be built here, in either direction. The codegen
module depends only on ignite-commons, while Message lives in ignite-nio, so
`Class<? extends Message>` does not even compile in this module. And the
classes themselves are unreachable regardless of the bound: every excluded one
lives in a module that depends on ignite-codegen (core, zookeeper, indexing,
calcite), so codegen cannot depend back — that would be a Maven cycle. This is
why the lists are FQN strings resolved through Elements.getTypeElement() with a
null filter, so an entry missing from the current module's classpath is simply
skipped. The mechanism is master's own; this PR adds a single entry to it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
