anton-vinogradov commented on code in PR #13095: URL: https://github.com/apache/ignite/pull/13095#discussion_r3664372207
########## modules/core/src/test/java/org/apache/ignite/internal/managers/communication/GridIoManagerOrderedPoisonMessageTest.java: ########## @@ -0,0 +1,247 @@ +/* + * 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.managers.communication; + +import java.util.Map; +import java.util.Queue; +import java.util.UUID; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.CoreMessagesProvider; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.processors.cache.CacheObjectContext; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.plugin.AbstractTestPluginProvider; +import org.apache.ignite.plugin.ExtensionRegistry; +import org.apache.ignite.plugin.PluginContext; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider; +import org.apache.ignite.plugin.extensions.communication.MessageMarshaller; +import org.apache.ignite.plugin.extensions.communication.MessageReader; +import org.apache.ignite.plugin.extensions.communication.MessageSerializer; +import org.apache.ignite.plugin.extensions.communication.MessageWriter; +import org.apache.ignite.testframework.GridTestUtils; +import org.apache.ignite.testframework.ListeningTestLogger; +import org.apache.ignite.testframework.LogListener; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.jetbrains.annotations.Nullable; +import org.junit.Test; + +import static org.apache.ignite.internal.managers.communication.GridIoPolicy.PUBLIC_POOL; + +/** + * An ordered message whose payload fails to unmarshal must be skipped with an error logged, not abandon the rest of Review Comment: Right, "ordered" is overloaded here. Added to the class javadoc: messages sent via GridIoManager#sendOrderedMessage are delivered in the order of sending — the receiver accumulates them per topic in a GridCommunicationMessageSet and drains the set in one go (unwind) — and that has nothing to do with the @Order annotation of the direct protocol. The test is about that drain: a message that fails to unmarshal must not abandon the rest of the set. ########## modules/core/src/test/java/org/apache/ignite/internal/managers/communication/GridIoManagerOrderedPoisonMessageTest.java: ########## @@ -0,0 +1,247 @@ +/* + * 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.managers.communication; + +import java.util.Map; +import java.util.Queue; +import java.util.UUID; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.CoreMessagesProvider; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.processors.cache.CacheObjectContext; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.plugin.AbstractTestPluginProvider; +import org.apache.ignite.plugin.ExtensionRegistry; +import org.apache.ignite.plugin.PluginContext; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider; +import org.apache.ignite.plugin.extensions.communication.MessageMarshaller; +import org.apache.ignite.plugin.extensions.communication.MessageReader; +import org.apache.ignite.plugin.extensions.communication.MessageSerializer; +import org.apache.ignite.plugin.extensions.communication.MessageWriter; +import org.apache.ignite.testframework.GridTestUtils; +import org.apache.ignite.testframework.ListeningTestLogger; +import org.apache.ignite.testframework.LogListener; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.jetbrains.annotations.Nullable; +import org.junit.Test; + +import static org.apache.ignite.internal.managers.communication.GridIoPolicy.PUBLIC_POOL; + +/** + * An ordered message whose payload fails to unmarshal must be skipped with an error logged, not abandon the rest of + * the accumulated set: the tail would otherwise stay unprocessed until the next message arrives on the topic. + */ +public class GridIoManagerOrderedPoisonMessageTest extends GridCommonAbstractTest { Review Comment: Agreed — renamed. The class is GridIoManagerOrderedUnmarshalFailureTest now, the nested ones are FailingUnmarshalMessage / FailingMarshaller, and the test method is testFailedMessageSkippedTailDelivered. No "poison" left in the file. ########## modules/core/src/test/java/org/apache/ignite/internal/managers/communication/GridIoManagerOrderedPoisonMessageTest.java: ########## @@ -0,0 +1,247 @@ +/* + * 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.managers.communication; + +import java.util.Map; +import java.util.Queue; +import java.util.UUID; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.CoreMessagesProvider; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.processors.cache.CacheObjectContext; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.plugin.AbstractTestPluginProvider; +import org.apache.ignite.plugin.ExtensionRegistry; +import org.apache.ignite.plugin.PluginContext; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider; +import org.apache.ignite.plugin.extensions.communication.MessageMarshaller; +import org.apache.ignite.plugin.extensions.communication.MessageReader; +import org.apache.ignite.plugin.extensions.communication.MessageSerializer; +import org.apache.ignite.plugin.extensions.communication.MessageWriter; +import org.apache.ignite.testframework.GridTestUtils; +import org.apache.ignite.testframework.ListeningTestLogger; +import org.apache.ignite.testframework.LogListener; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.jetbrains.annotations.Nullable; +import org.junit.Test; + +import static org.apache.ignite.internal.managers.communication.GridIoPolicy.PUBLIC_POOL; + +/** + * An ordered message whose payload fails to unmarshal must be skipped with an error logged, not abandon the rest of + * the accumulated set: the tail would otherwise stay unprocessed until the next message arrives on the topic. + */ +public class GridIoManagerOrderedPoisonMessageTest extends GridCommonAbstractTest { + /** Direct type for the test message, past the core range. */ + private static final short TYPE = (short)(CoreMessagesProvider.MAX_MESSAGE_ID + 10); + + /** Ordered topic under test. */ + private static final String TOPIC = "poison-test-topic"; Review Comment: Renamed to `unmarshal-failure-test-topic` (and the plugin name to `ordered-unmarshal-failure-test`). ########## modules/core/src/test/java/org/apache/ignite/internal/managers/communication/GridIoManagerOrderedPoisonMessageTest.java: ########## @@ -0,0 +1,247 @@ +/* + * 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.managers.communication; + +import java.util.Map; +import java.util.Queue; +import java.util.UUID; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.CoreMessagesProvider; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.processors.cache.CacheObjectContext; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.plugin.AbstractTestPluginProvider; +import org.apache.ignite.plugin.ExtensionRegistry; +import org.apache.ignite.plugin.PluginContext; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider; +import org.apache.ignite.plugin.extensions.communication.MessageMarshaller; +import org.apache.ignite.plugin.extensions.communication.MessageReader; +import org.apache.ignite.plugin.extensions.communication.MessageSerializer; +import org.apache.ignite.plugin.extensions.communication.MessageWriter; +import org.apache.ignite.testframework.GridTestUtils; +import org.apache.ignite.testframework.ListeningTestLogger; +import org.apache.ignite.testframework.LogListener; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.jetbrains.annotations.Nullable; +import org.junit.Test; + +import static org.apache.ignite.internal.managers.communication.GridIoPolicy.PUBLIC_POOL; + +/** + * An ordered message whose payload fails to unmarshal must be skipped with an error logged, not abandon the rest of + * the accumulated set: the tail would otherwise stay unprocessed until the next message arrives on the topic. + */ +public class GridIoManagerOrderedPoisonMessageTest extends GridCommonAbstractTest { Review Comment: I'd keep it separate. The test needs two grids, a plugin registering its own direct type with a marshaller that fails on demand, and a listener thread blocked mid-delivery so that the next two messages pile up in one ordered set — that is the whole point of the scenario. The neighbours around serialization/marshalling are unit-level: MessageProcessorTest (compile-testing), MarshallerCacheFreeUnmarshalTest and MessageSerializationArchitectureTest (class-graph rules), MessageUnmarshalOnceCheckTest — none of them start a node. The only grid-based one is MessageMarshalOnceTest, but its configuration installs a different plugin (two direct types with counting marshallers) plus an SPI that fails the first send; folding a third direct type and a blocking listener into it would merge two unrelated scenarios into one configuration. That is also the reasoning you applied yourself in the MessageSerializationArchitectureTest thread — different subjects, different classes. ########## modules/core/src/test/java/org/apache/ignite/internal/managers/communication/GridIoManagerOrderedUnmarshalFailureTest.java: ########## @@ -0,0 +1,247 @@ +/* + * 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.managers.communication; + +import java.util.Map; +import java.util.Queue; +import java.util.UUID; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.CoreMessagesProvider; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.processors.cache.CacheObjectContext; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.plugin.AbstractTestPluginProvider; +import org.apache.ignite.plugin.ExtensionRegistry; +import org.apache.ignite.plugin.PluginContext; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider; +import org.apache.ignite.plugin.extensions.communication.MessageMarshaller; +import org.apache.ignite.plugin.extensions.communication.MessageReader; +import org.apache.ignite.plugin.extensions.communication.MessageSerializer; +import org.apache.ignite.plugin.extensions.communication.MessageWriter; +import org.apache.ignite.testframework.GridTestUtils; +import org.apache.ignite.testframework.ListeningTestLogger; +import org.apache.ignite.testframework.LogListener; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.jetbrains.annotations.Nullable; +import org.junit.Test; + +import static org.apache.ignite.internal.managers.communication.GridIoPolicy.PUBLIC_POOL; + +/** + * An ordered message whose payload fails to unmarshal must be skipped with an error logged, not abandon the rest of + * the accumulated set: the tail would otherwise stay unprocessed until the next message arrives on the topic. + */ +public class GridIoManagerOrderedPoisonMessageTest extends GridCommonAbstractTest { + /** Direct type for the test message, past the core range. */ + private static final short TYPE = (short)(CoreMessagesProvider.MAX_MESSAGE_ID + 10); + + /** Ordered topic under test. */ + private static final String TOPIC = "poison-test-topic"; + + /** Listener log of the receiver. */ + private ListeningTestLogger lsnrLog; + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName); + + if (getTestIgniteInstanceIndex(igniteInstanceName) == 1) + cfg.setGridLogger(lsnrLog); + + cfg.setPluginProviders(new AbstractTestPluginProvider() { + @Override public String name() { + return "ordered-poison-test"; + } + + @Override public void initExtensions(PluginContext ctx, ExtensionRegistry registry) { + registry.registerExtension(MessageFactoryProvider.class, factory -> + factory.register(TYPE, PoisonableMessage::new, new Serializer(), new PoisoningMarshaller())); + } + }); + + return cfg; + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + stopAllGrids(); + + super.afterTest(); + } + + /** A poisoned message between two valid ones must not prevent the delivery of the one behind it. */ + @Test + public void testPoisonedMessageSkippedTailDelivered() throws Exception { + lsnrLog = new ListeningTestLogger(log); + + LogListener skipLsnr = LogListener.matches("Failed to unmarshal ordered message (will skip)").build(); + + lsnrLog.registerListener(skipLsnr); + + IgniteEx snd = startGrid(0); + IgniteEx rcv = startGrid(1); + + Queue<Integer> delivered = new ConcurrentLinkedQueue<>(); + CountDownLatch firstReceived = new CountDownLatch(1); + CountDownLatch proceed = new CountDownLatch(1); + CountDownLatch tailReceived = new CountDownLatch(1); + + rcv.context().io().addMessageListener(TOPIC, (nodeId, msg, plc) -> { + PoisonableMessage m = (PoisonableMessage)msg; + + delivered.add(m.seq); + + if (m.seq == 1) { + firstReceived.countDown(); + + U.awaitQuiet(proceed); + } + else + tailReceived.countDown(); + }); + + GridIoManager sndIo = snd.context().io(); + + // First message occupies the striped worker in the listener above. + sndIo.sendOrderedMessage(rcv.localNode(), TOPIC, new PoisonableMessage(1, false), PUBLIC_POOL, 30_000, false); + + assertTrue("First message must be delivered", firstReceived.await(10, TimeUnit.SECONDS)); + + // While the worker is blocked, the poisoned message and the one behind it accumulate in the same set. + sndIo.sendOrderedMessage(rcv.localNode(), TOPIC, new PoisonableMessage(2, true), PUBLIC_POOL, 30_000, false); + sndIo.sendOrderedMessage(rcv.localNode(), TOPIC, new PoisonableMessage(3, false), PUBLIC_POOL, 30_000, false); + + assertTrue("Both messages must reach the receiver's ordered set before the worker resumes", + GridTestUtils.waitForCondition(() -> pendingOrderedMessages(rcv) == 2, 10_000)); + + proceed.countDown(); + + assertTrue("The message behind the poisoned one must still be delivered", tailReceived.await(10, TimeUnit.SECONDS)); + + assertEquals("[1, 3]", delivered.toString()); + assertTrue("Skip must be logged", skipLsnr.check(10_000)); + } + + /** @return Number of messages pending in the receiver's ordered set of {@link #TOPIC}. */ + private static int pendingOrderedMessages(IgniteEx rcv) { + Map<Object, Map<UUID, Object>> setMap = GridTestUtils.getFieldValue(rcv.context().io(), "msgSetMap"); Review Comment: That was the shape this test had at first, and it was rejected in review: initMessageFactoryForTest() in IgniteKernal was removed on the grounds that a production class should not carry a method existing solely for one test, and the test was switched to reflection (U.invoke) instead — see the IgniteKernal thread. Adding a @TestOnly getter to GridIoManager for msgSetMap would walk that decision back, so I'd rather keep the reflection here: it is a test-only peek at internal state, contained in the test itself. Happy to switch if you and @shishkovilja agree on the opposite rule. -- 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]
