You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by am...@apache.org on 2019/04/09 13:22:45 UTC
[ignite] branch master updated: IGNITE-11460: MVCC: Possible race
on coordinator changing on client reconnection. This closes #6211.
This is an automated email from the ASF dual-hosted git repository.
amashenkov 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 dce279b IGNITE-11460: MVCC: Possible race on coordinator changing on client reconnection. This closes #6211.
dce279b is described below
commit dce279b74ca81524e882445e47afdeb2c7c0aa96
Author: Nikita Amelchev <ns...@gmail.com>
AuthorDate: Tue Apr 9 16:22:31 2019 +0300
IGNITE-11460: MVCC: Possible race on coordinator changing on client reconnection. This closes #6211.
Signed-off-by: Andrey V. Mashenkov <an...@gmail.com>
---
.../processors/cache/mvcc/MvccProcessorImpl.java | 24 +++--
.../cache/mvcc/CacheMvccClientReconnectTest.java | 111 +++++++++++++++++++++
.../testsuites/IgniteCacheMvccTestSuite.java | 4 +-
3 files changed, 131 insertions(+), 8 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessorImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessorImpl.java
index 7e32b59..234c2e8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessorImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/mvcc/MvccProcessorImpl.java
@@ -239,6 +239,9 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
/** */
private final CustomEventListener customLsnr;
+ /** State mutex. */
+ private final Object stateMux = new Object();
+
/**
* @param ctx Context.
*/
@@ -439,7 +442,9 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
// Notify all listeners waiting for a snapshot.
onCoordinatorFailed(curCrd0.nodeId());
- curCrd = MvccCoordinator.DISCONNECTED_COORDINATOR;
+ synchronized (stateMux) {
+ curCrd = MvccCoordinator.DISCONNECTED_COORDINATOR;
+ }
readyVer = AffinityTopologyVersion.NONE;
}
@@ -554,15 +559,20 @@ public class MvccProcessorImpl extends GridProcessorAdapter implements MvccProce
private void onCoordinatorChanged(AffinityTopologyVersion topVer, Collection<ClusterNode> nodes, boolean sndQrys) {
MvccCoordinator newCrd = pickMvccCoordinator(nodes, topVer);
- if (newCrd.disconnected()) {
- curCrd = newCrd;
+ synchronized (stateMux) {
+ if (ctx.clientDisconnected())
+ return;
- return;
- }
+ if (newCrd.disconnected()) {
+ curCrd = newCrd;
- assert newCrd.topologyVersion().compareTo(curCrd.topologyVersion()) > 0;
+ return;
+ }
+
+ assert newCrd.topologyVersion().compareTo(curCrd.topologyVersion()) > 0;
- curCrd = newCrd;
+ curCrd = newCrd;
+ }
processActiveQueries(nodes, newCrd, sndQrys);
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccClientReconnectTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccClientReconnectTest.java
new file mode 100644
index 0000000..9329814
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccClientReconnectTest.java
@@ -0,0 +1,111 @@
+/*
+ * 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.mvcc;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.events.Event;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.lang.IgnitePredicate;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.Test;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.ignite.events.EventType.EVT_NODE_JOINED;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator.DISCONNECTED_COORDINATOR;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+
+/**
+ * Tests Mvcc coordinator change on client reconnect.
+ */
+public class CacheMvccClientReconnectTest extends GridCommonAbstractTest {
+ /** */
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ if (igniteInstanceName.contains("client")) {
+ cfg.setClientMode(true);
+
+ Map<IgnitePredicate<? extends Event>, int[]> lsnrs = new HashMap<>();
+
+ lsnrs.put(new IgnitePredicate<Event>() {
+ @Override public boolean apply(Event evt) {
+ try {
+ // Wait for the discovery notifier worker processed client disconnection.
+ latch.await();
+ }
+ catch (InterruptedException e) {
+ log.error("Unexpected exception.", e);
+
+ fail("Unexpected exception: " + e.getMessage());
+ }
+
+ return true;
+ }
+ }, new int[] {EVT_NODE_JOINED});
+
+ cfg.setLocalEventListeners(lsnrs);
+ }
+
+ return cfg;
+ }
+
+ /**
+ * Checks that events processed after client disconnect will not change coordinator until client reconnected.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testClientReconnect() throws Exception {
+ startGrid(0);
+
+ IgniteEx client = startGrid("client");
+
+ MvccProcessor coordProc = client.context().coordinators();
+
+ // Creates the join event.
+ startGrid(1);
+
+ stopGrid(0, true);
+ stopGrid(1, true);
+
+ client.context().discovery().reconnect();
+
+ // Wait for the discovery notifier worker processed client disconnection.
+ assertTrue("Failed to wait for client disconnected.",
+ waitForCondition(() -> client.cluster().clientReconnectFuture() != null, 10_000));
+
+ assertTrue("Failed to wait for setting disconnected coordinator.", waitForCondition(
+ () -> DISCONNECTED_COORDINATOR.equals(coordProc.currentCoordinator()), 2000));
+
+ // The discovery event thread may continue processing events when the notifier worker already processed
+ // the client disconnection or a local join. It may lead to setting a wrong coordinator.
+ latch.countDown();
+
+ startGrid(0);
+
+ client.cluster().clientReconnectFuture().get(10, SECONDS);
+
+ assertEquals(grid(0).localNode().id(), coordProc.currentCoordinator().nodeId());
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheMvccTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheMvccTestSuite.java
index 75d7174..b43bd33 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheMvccTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheMvccTestSuite.java
@@ -17,6 +17,7 @@
package org.apache.ignite.testsuites;
+import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccClientReconnectTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccClusterRestartTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccConfigurationValidationTest;
import org.apache.ignite.internal.processors.cache.mvcc.CacheMvccIteratorWithConcurrentTransactionTest;
@@ -68,7 +69,8 @@ import org.junit.runners.Suite;
CacheMvccClusterRestartTest.class,
CacheMvccPartitionedCoordinatorFailoverTest.class,
CacheMvccReplicatedCoordinatorFailoverTest.class,
- CacheMvccProcessorLazyStartTest.class
+ CacheMvccProcessorLazyStartTest.class,
+ CacheMvccClientReconnectTest.class
})
public class IgniteCacheMvccTestSuite {
}