Vladsz83 commented on code in PR #13095:
URL: https://github.com/apache/ignite/pull/13095#discussion_r3659381754
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryBatchMessage.java:
##########
@@ -61,7 +62,10 @@ public QueryBatchMessage(UUID qryId, long fragmentId, long
exchangeId, int batch
this.batchId = batchId;
this.last = last;
- mRows = rows.stream().map(o -> o == null ? null : new
GenericValueMessage(o)).collect(Collectors.toList());
+ mRows = new ArrayList<>(rows.size());
Review Comment:
Why changed?
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryBatchMessage.java:
##########
@@ -99,6 +103,11 @@ public boolean last() {
* @return Rows.
*/
public List<Object> rows() {
- return
mRows.stream().map(GenericValueMessage::value).collect(Collectors.toList());
+ List<Object> rows = new ArrayList<>(mRows.size());
Review Comment:
Same
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryBatchMessage.java:
##########
@@ -61,7 +62,10 @@ public QueryBatchMessage(UUID qryId, long fragmentId, long
exchangeId, int batch
this.batchId = batchId;
this.last = last;
- mRows = rows.stream().map(o -> o == null ? null : new
GenericValueMessage(o)).collect(Collectors.toList());
+ mRows = new ArrayList<>(rows.size());
Review Comment:
Why changed?
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryBatchMessage.java:
##########
@@ -99,6 +103,11 @@ public boolean last() {
* @return Rows.
*/
public List<Object> rows() {
- return
mRows.stream().map(GenericValueMessage::value).collect(Collectors.toList());
+ List<Object> rows = new ArrayList<>(mRows.size());
Review Comment:
Same
##########
modules/codegen/src/main/java/org/apache/ignite/internal/MessageGenerator.java:
##########
@@ -0,0 +1,289 @@
+/*
+ * 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;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.PrintWriter;
+import java.io.Reader;
+import java.io.Writer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+import javax.annotation.processing.FilerException;
+import javax.annotation.processing.ProcessingEnvironment;
+import javax.lang.model.element.TypeElement;
+import javax.lang.model.element.VariableElement;
+import javax.lang.model.type.TypeMirror;
+import javax.lang.model.util.ElementFilter;
+import javax.lang.model.util.Elements;
+import javax.tools.Diagnostic;
+import javax.tools.FileObject;
+import javax.tools.JavaFileObject;
+import javax.tools.StandardLocation;
+import
org.apache.ignite.internal.systemview.SystemViewRowAttributeWalkerProcessor;
+
+/**
+ * Base class for message code generators ({@link MessageSerializerGenerator},
{@link MessageMarshallerGenerator},
+ * {@link MessageDeploymentGenerator}).
+ */
+public abstract class MessageGenerator {
Review Comment:
Sounds like something that generates messages. By a template maybe.
Actually, it produses various message processors like Deployer or Serializatior.
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxRecreateEvictionTest.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.cache.distributed.near;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
+import
org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareRequest;
+import
org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareResponse;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+import org.junit.Test;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+import static org.apache.ignite.transactions.TransactionConcurrency.OPTIMISTIC;
+import static
org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
+
+/**
+ * Checks that a near-write {@link GridDhtTxPrepareRequest} targeting a cache
incarnation older than the receiver's
+ * current one is answered with the key in {@code nearEvicted}, so the reader
is dropped instead of keeping a stale
+ * near entry.
+ */
+public class GridNearTxRecreateEvictionTest extends GridCommonAbstractTest {
Review Comment:
Can we put in other ticket?
##########
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:
Not ignite-style name I believe. We have tons of failure simulation in the
tests. But we do not use word `poison`. Let's rename it.
##########
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:
Same: `failure-test-topic`?
##########
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:
Why we keep creating test class for single test method? Let's put it
somewhere aroud serialization/marshalling tests
##########
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:
What does it mean `ordered message`? What order? `@Order`? We might need
some links here.
##########
modules/codegen/src/main/java/org/apache/ignite/internal/MessageGenerator.java:
##########
@@ -0,0 +1,289 @@
+/*
+ * 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;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.PrintWriter;
+import java.io.Reader;
+import java.io.Writer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+import javax.annotation.processing.FilerException;
+import javax.annotation.processing.ProcessingEnvironment;
+import javax.lang.model.element.TypeElement;
+import javax.lang.model.element.VariableElement;
+import javax.lang.model.type.TypeMirror;
+import javax.lang.model.util.ElementFilter;
+import javax.lang.model.util.Elements;
+import javax.tools.Diagnostic;
+import javax.tools.FileObject;
+import javax.tools.JavaFileObject;
+import javax.tools.StandardLocation;
+import
org.apache.ignite.internal.systemview.SystemViewRowAttributeWalkerProcessor;
+
+/**
+ * Base class for message code generators ({@link MessageSerializerGenerator},
{@link MessageMarshallerGenerator},
+ * {@link MessageDeploymentGenerator}).
+ */
+public abstract class MessageGenerator {
+ /** Blank separator line in generated code. */
+ public static final String EMPTY = "";
+
+ /** Platform line separator used in generated code. */
+ public static final String NL = System.lineSeparator();
+
+ /** Single indentation unit. */
+ public static final String TAB = " ";
+
+ /** Javadoc stub emitted on every generated {@code @Override} method. */
+ public static final String METHOD_JAVADOC = "/** */";
+
+ /** Javadoc block emitted on every generated class. */
+ public static final String CLS_JAVADOC = "/**" + NL +
+ " * This class is generated automatically." + NL +
+ " *" + NL +
+ " * @see " + MessageProcessor.class.getName() + NL +
+ " */";
+
+ /** */
+ protected final ProcessingEnvironment env;
+
+ /** */
+ protected final Set<String> imports = new TreeSet<>();
+
+ /** */
+ protected TypeElement type;
+
+ /** Current indentation level. Set to the class-member level once in
{@link #generate}; adjusted only by balanced shifts. */
+ protected int indent;
+
+ /** Dispatches cache-object-context resolution in both the marshaller and
the deployment generators. */
+ protected final TypeMirror cacheIdAwareType;
+
+ /** */
+ MessageGenerator(ProcessingEnvironment env) {
+ this.env = env;
+
+ cacheIdAwareType =
type("org.apache.ignite.plugin.extensions.communication.CacheIdAware");
+ }
+
+ /** */
+ boolean isCacheIdAwareMessage(TypeElement te) {
+ return assignableFrom(te.asType(), cacheIdAwareType);
+ }
+
+ /** Generates and writes the source file for {@code type}; skipped when
{@link #shouldSkip} returns {@code true}. */
+ final void generate(TypeElement type, List<VariableElement> fields) throws
Exception {
+ assert this.type == null : "Message" + typeSuffix() + " generator
isn't stateless and is supposed to be single-use.";
+
+ if (shouldSkip(type, fields))
+ return;
+
+ this.type = type;
+
+ indent = 1;
+
+ generateBody(fields);
+
+ String clsName = type.getSimpleName() + typeSuffix();
+ String fqnClsName = env.getElementUtils().getPackageOf(type) + "." +
clsName;
+ String code = buildClassCode(clsName);
+
+ if (code == null)
+ return;
+
+ try {
+ JavaFileObject file = env.getFiler().createSourceFile(fqnClsName);
+
+ try (Writer writer = file.openWriter()) {
+ writer.append(code);
+ writer.flush();
+ }
+ }
+ catch (FilerException e) {
+ if (!identicalFileIsAlreadyGenerated(env, code, fqnClsName)) {
+ env.getMessager().printMessage(Diagnostic.Kind.ERROR,
+ "Message" + typeSuffix() + " " + clsName + " is already
generated. Try 'mvn clean install' to fix the issue.");
+
+ throw e;
+ }
+ }
+ }
+
+ /** @return Class name suffix: {@code "Serializer"} or {@code
"Marshaller"}. */
+ abstract String typeSuffix();
+
+ /** @return {@code true} if no file should be generated for this type;
default is {@code false}. */
+ boolean shouldSkip(TypeElement type, List<VariableElement> fields) {
Review Comment:
Is only for subclasses. `protected` maybe? The others too.
##########
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";
+
+ /** 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:
We might cat to the implementation and add some getter with `@TestOnly`
--
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]
