timoninmaxim commented on code in PR #11855:
URL: https://github.com/apache/ignite/pull/11855#discussion_r1991061911
##########
modules/core/src/main/java/org/apache/ignite/internal/client/thin/ReliableChannel.java:
##########
@@ -678,7 +675,7 @@ synchronized void initChannelHolders() {
int idx = curChIdx;
- if (idx != -1)
+ if (idx != -1 && holders != null)
Review Comment:
Looks like, that in case when `idx != -1` then `holders` never `null`. Could
you please describe the scenario?
##########
modules/core/src/test/java/org/apache/ignite/client/ReliabilityTest.java:
##########
@@ -259,42 +265,67 @@ public void testRetryReadPolicyRetriesCacheGet() {
// Fail.
dropAllThinClientConnections(Ignition.allGrids().get(0));
- // Reuse second address without fail.
+ if (!partitionAware) {
+ Throwable ex = GridTestUtils.assertThrowsWithCause(() ->
cache.get(0), ClientConnectionException.class);
+
+ GridTestUtils.assertContains(null, ex.getMessage(),
F.first(cluster.clientAddresses()));
+ }
+
+ // Reuse the address after recovery without fail.
cache.get(0);
}
}
/**
- * Tests retry policy exception handling.
+ * Tests that retry policy exception handling for async operations
propagates to the caller.
*/
@Test
- public void testExceptionInRetryPolicyPropagatesToCaller() {
+ public void testExceptionInRetryPolicyPropagatesToCallerAsync() {
Assume.assumeFalse(partitionAware);
try (LocalIgniteCluster cluster = LocalIgniteCluster.start(1);
IgniteClient client =
Ignition.startClient(getClientConfiguration()
.setRetryPolicy(new ExceptionRetryPolicy())
- .setAddresses(F.first(cluster.clientAddresses()),
F.first(cluster.clientAddresses()))
+ .setAddresses(F.first(cluster.clientAddresses()))
.setClusterDiscoveryEnabled(false))
) {
ClientCache<Integer, Integer> cache = client.createCache("cache");
+
dropAllThinClientConnections(Ignition.allGrids().get(0));
- Throwable asyncEx = GridTestUtils.assertThrows(null, () ->
cache.getAsync(0).get(),
- ExecutionException.class, "Channel is closed");
+ Throwable asyncEx = GridTestUtils.assertThrows(null,
+ () -> cache.getAsync(0).get(),
+ ExecutionException.class, "Channel is closed");
GridTestUtils.assertContains(null, asyncEx.getMessage(),
F.first(cluster.clientAddresses()));
+ }
+ }
+
+
+ /**
+ * Tests that retry policy exception handling for sync operations
propagates to the caller.
+ */
+ @Test
+ public void testExceptionInRetryPolicyPropagatesToCallerSync() {
+ Assume.assumeFalse(partitionAware);
+
+ try (LocalIgniteCluster cluster = LocalIgniteCluster.start(1);
+ IgniteClient client =
Ignition.startClient(getClientConfiguration()
+ .setRetryPolicy(new ExceptionRetryPolicy())
+ .setAddresses(F.first(cluster.clientAddresses()))
+ .setClusterDiscoveryEnabled(false))
+ ) {
+ ClientCache<Integer, Integer> cache = client.createCache("cache");
dropAllThinClientConnections(Ignition.allGrids().get(0));
- Throwable syncEx = GridTestUtils.assertThrows(null, () ->
cache.get(0),
+ Throwable syncEx = GridTestUtils.assertThrows(null,
Review Comment:
useless change
##########
modules/core/src/main/java/org/apache/ignite/internal/client/thin/ReliableChannel.java:
##########
@@ -918,8 +919,22 @@ private <T> T applyOnNodeChannelWithFallback(UUID
tryNodeId, Function<ClientChan
try {
channel = hld.getOrCreateChannel();
+ try {
Review Comment:
Add empty line before try
##########
modules/core/src/test/java/org/apache/ignite/internal/client/thin/ReliableChannelDuplicationTest.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.client.thin;
+
+import java.net.InetSocketAddress;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.junit.Assume;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static java.util.stream.IntStream.range;
+
+/**
+ * Tests for duplication in channels' list.
+ */
+@RunWith(Parameterized.class)
+public class ReliableChannelDuplicationTest extends
ThinClientAbstractPartitionAwarenessTest {
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ stopAllGrids();
+ }
+
+ /** Grid count. */
+ @Parameterized.Parameter(0)
+ public int gridCnt;
+
+ /** */
+ @Parameterized.Parameters(name = "gridCount = {0}")
+ public static Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] {
+ { 1 },
+ { 3 }
+ });
+ }
+
+ /**
+ * Test after cluster restart the number of channels remains equal to the
number of nodes.
+ */
+ @Test
+ public void testDuplicationOnClusterRestart() throws Exception {
+ startGrids(gridCnt);
+
+ initClient(getClientConfiguration(range(0, gridCnt).toArray()),
range(0, gridCnt).toArray());
+
+
assertNoDuplicates(((TcpIgniteClient)client).reliableChannel().getChannelHolders());
+
+ stopAllGrids();
+
+ startGrids(gridCnt);
+
+ client.cacheNames();
+
+
assertNoDuplicates(((TcpIgniteClient)client).reliableChannel().getChannelHolders());
+ }
+
+ /**
+ * Test behavior after stopping a single node in the cluster.
+ */
+ @Test
+ public void testStopSingleNodeDuringOperation() throws Exception {
+ Assume.assumeFalse(gridCnt == 1);
+
+ testChannelDuplication(gridCnt, 1, 0);
+ }
+
+ /**
+ * Test behavior after stopping and restarting a node.
+ */
+ @Test
+ public void testStopAndRestartNode() throws Exception {
+ Assume.assumeFalse(gridCnt == 1);
+
+ testChannelDuplication(gridCnt, 1, 1);
+ }
+
+ /**
+ * Test behavior after stopping multiple nodes in the cluster.
+ */
+ @Test
+ public void testStopMultipleNodesDuringOperation() throws Exception {
+ Assume.assumeFalse(gridCnt < 3);
+
+ testChannelDuplication(gridCnt, 2, 2);
+ }
+
+ /**
+ * Asserts that there are no duplicate channels in the list of holders
based on their remote addresses.
+ *
+ * @param holders List of channel holders.
+ */
+ private void assertNoDuplicates(List<ReliableChannel.ClientChannelHolder>
holders) {
+ Set<InetSocketAddress> addrs = new HashSet<>();
+
+ for (ReliableChannel.ClientChannelHolder holder : holders) {
+ holder.getAddresses().forEach(addr -> {
+ if (!addrs.add(addr))
+ throw new AssertionError("Duplicate remote address found:
" + addr);
+ });
+ }
+ }
+
+ /**
+ * Stop a Node and provide an operation to notify the client about new
topology.
+ */
+ private void stopNodeAndMakeTopologyChangeDetection(int idx) {
+ stopGrid(idx);
+
+ detectTopologyChange();
+
+ client.cacheNames();
+ }
+
+ /**
+ * Tests that no duplicate channel holders are created during node
restarts and topology changes.
+ *
+ * @param gridCnt int Grids to start.
+ * @param gridsStop int Grids to stop.
+ * @param gridsRestart int Grids to restart after stop.
+ */
+ private void testChannelDuplication(int gridCnt, int gridsStop, int
gridsRestart) throws Exception {
Review Comment:
`gridCnt` is already a class level param.
##########
modules/core/src/main/java/org/apache/ignite/internal/client/thin/ReliableChannel.java:
##########
@@ -931,7 +946,6 @@ private <T> T applyOnNodeChannelWithFallback(UUID
tryNodeId, Function<ClientChan
throw e;
}
}
-
Review Comment:
Remove useless change
##########
modules/core/src/main/java/org/apache/ignite/internal/client/thin/ReliableChannel.java:
##########
@@ -918,8 +919,22 @@ private <T> T applyOnNodeChannelWithFallback(UUID
tryNodeId, Function<ClientChan
try {
channel = hld.getOrCreateChannel();
+ try {
+ return function.apply(channel);
+ }
+ catch (ClientConnectionException e) {
+ if (partitionAwarenessEnabled) {
Review Comment:
Why do we care about `partitionAwarenessEnabled` here?
##########
modules/core/src/test/java/org/apache/ignite/internal/client/thin/ReliableChannelDuplicationTest.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.client.thin;
+
+import java.net.InetSocketAddress;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.junit.Assume;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static java.util.stream.IntStream.range;
+
+/**
+ * Tests for duplication in channels' list.
+ */
+@RunWith(Parameterized.class)
+public class ReliableChannelDuplicationTest extends
ThinClientAbstractPartitionAwarenessTest {
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ stopAllGrids();
+ }
+
+ /** Grid count. */
+ @Parameterized.Parameter(0)
+ public int gridCnt;
+
+ /** */
+ @Parameterized.Parameters(name = "gridCount = {0}")
+ public static Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] {
+ { 1 },
+ { 3 }
+ });
+ }
+
+ /**
+ * Test after cluster restart the number of channels remains equal to the
number of nodes.
+ */
+ @Test
+ public void testDuplicationOnClusterRestart() throws Exception {
+ startGrids(gridCnt);
+
+ initClient(getClientConfiguration(range(0, gridCnt).toArray()),
range(0, gridCnt).toArray());
+
+
assertNoDuplicates(((TcpIgniteClient)client).reliableChannel().getChannelHolders());
+
+ stopAllGrids();
+
+ startGrids(gridCnt);
+
+ client.cacheNames();
Review Comment:
Which guarantee do you want get from this call?
--
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: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
