This is an automated email from the ASF dual-hosted git repository.
ppa 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 6dcfbd9ee5f IGNITE-16805 Fixed an issue with stopping a node during
partition map exchange (#9945)
6dcfbd9ee5f is described below
commit 6dcfbd9ee5faa26240f187b2ef23b7874e362038
Author: Pavel Pereslegin <xxt...@gmail.com>
AuthorDate: Wed Apr 13 11:01:01 2022 +0300
IGNITE-16805 Fixed an issue with stopping a node during partition map
exchange (#9945)
---
.../cache/GridCachePartitionExchangeManager.java | 7 +-
.../preloader/GridDhtPartitionsExchangeFuture.java | 3 +
.../GridCacheAbstractNodeRestartSelfTest.java | 435 ++++++++++-----------
3 files changed, 203 insertions(+), 242 deletions(-)
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
index 6a06e8c6483..62d937f537f 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
@@ -1888,10 +1888,11 @@ public class GridCachePartitionExchangeManager<K, V>
extends GridCacheSharedMana
if (initTopVer.compareTo(fut.exchangeId().topologyVersion()) <
0)
continue;
- skipped++;
-
- if (skipped > 10)
+ // Skip recent exchange futures (ignore futures that have been
merged into another).
+ if (skipped == 10 || fut.isMerged())
fut.cleanUp();
+ else
+ skipped++;
}
}
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
index 3e49fbfbf66..ce27be0e652 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
@@ -5520,6 +5520,9 @@ public class GridDhtPartitionsExchangeFuture extends
GridDhtTopologyFutureAdapte
delayedLatestMsg = fullMsg;
listen(f -> {
+ if (f.error() != null)
+ return;
+
GridDhtPartitionsFullMessage msg;
synchronized (this) {
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheAbstractNodeRestartSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheAbstractNodeRestartSelfTest.java
index 9a62cfcd33a..a2ff1d6cb31 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheAbstractNodeRestartSelfTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheAbstractNodeRestartSelfTest.java
@@ -18,10 +18,8 @@
package org.apache.ignite.internal.processors.cache.distributed;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashMap;
-import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Random;
@@ -39,8 +37,10 @@ import org.apache.ignite.cache.CacheRebalanceMode;
import org.apache.ignite.cache.eviction.lru.LruEvictionPolicy;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.util.future.GridCompoundFuture;
import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.GridTestUtils.SF;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.apache.ignite.transactions.Transaction;
@@ -590,111 +590,94 @@ public abstract class
GridCacheAbstractNodeRestartSelfTest extends GridCommonAbs
startGrids();
- Collection<Thread> threads = new LinkedList<>();
+ GridCompoundFuture<Long, Long> fut = new GridCompoundFuture<>();
try {
final AtomicInteger putCntr = new AtomicInteger();
-
final CyclicBarrier barrier = new CyclicBarrier(putThreads +
restartThreads);
+ final AtomicInteger startedThreadCntr = new AtomicInteger();
- for (int i = 0; i < putThreads; i++) {
- final int gridIdx = i;
-
- Thread t = new Thread(new Runnable() {
- @Override public void run() {
- try {
- barrier.await();
+ fut.add(GridTestUtils.runMultiThreadedAsync(() -> {
+ try {
+ int gridIdx = startedThreadCntr.getAndIncrement();
- info("Starting put thread: " + gridIdx);
+ barrier.await();
- Thread.currentThread().setName("put-worker-" +
grid(gridIdx).name());
+ info("Starting put thread: " + gridIdx);
- IgniteCache<Integer, String> cache =
grid(gridIdx).cache(CACHE_NAME);
+ Thread.currentThread().setName("put-worker-" +
grid(gridIdx).name());
- while (System.currentTimeMillis() < endTime &&
err.get() == null) {
- int key = RAND.nextInt(keyCnt);
+ IgniteCache<Integer, String> cache =
grid(gridIdx).cache(CACHE_NAME);
- try {
- cache.put(key, Integer.toString(key));
- }
- catch (IgniteException | CacheException
ignored) {
- // It is ok if primary node leaves grid.
- }
+ while (System.currentTimeMillis() < endTime && err.get()
== null) {
+ int key = RAND.nextInt(keyCnt);
- cache.get(key);
+ try {
+ cache.put(key, Integer.toString(key));
+ }
+ catch (IgniteException | CacheException ignored) {
+ // It is ok if primary node leaves grid.
+ }
- int c = putCntr.incrementAndGet();
+ cache.get(key);
- if (c % LOG_FREQ == 0)
- info(">>> Put iteration [cnt=" + c + ",
key=" + key + ']');
- }
- }
- catch (Exception e) {
- err.compareAndSet(null, e);
+ int c = putCntr.incrementAndGet();
- error("Unexpected exception in put-worker.", e);
- }
+ if (c % LOG_FREQ == 0)
+ info(">>> Put iteration [cnt=" + c + ", key=" +
key + ']');
}
- }, "put-worker-" + i);
-
- t.start();
+ }
+ catch (Exception e) {
+ err.compareAndSet(null, e);
- threads.add(t);
- }
+ error("Unexpected exception in put-worker.", e);
+ }
+ }, putThreads, "put-worker"));
- for (int i = 0; i < restartThreads; i++) {
- final int gridIdx = i + putThreads;
+ GridTestUtils.waitForCondition(() -> startedThreadCntr.get() ==
putThreads, getTestTimeout());
- Thread t = new Thread(new Runnable() {
- @Override public void run() {
- try {
- barrier.await();
+ fut.add(GridTestUtils.runMultiThreadedAsync(() -> {
+ try {
+ int gridIdx = startedThreadCntr.getAndIncrement();
- info("Starting restart thread: " + gridIdx);
+ barrier.await();
- int cnt = 0;
+ info("Starting restart thread: " + gridIdx);
- while (System.currentTimeMillis() < endTime &&
err.get() == null) {
- log.info(">>>>>>> Stopping grid " + gridIdx);
+ int cnt = 0;
- stopGrid(gridIdx);
+ while (System.currentTimeMillis() < endTime && err.get()
== null) {
+ log.info(">>>>>>> Stopping grid " + gridIdx);
- log.info(">>>>>>> Starting grid " + gridIdx);
+ stopGrid(gridIdx);
- startGrid(gridIdx);
+ log.info(">>>>>>> Starting grid " + gridIdx);
- int c = ++cnt;
+ startGrid(gridIdx);
- if (c % LOG_FREQ == 0)
- info(">>> Restart iteration: " + c);
- }
- }
- catch (Exception e) {
- err.compareAndSet(null, e);
+ int c = ++cnt;
- error("Unexpected exception in restart-worker.",
e);
- }
+ if (c % LOG_FREQ == 0)
+ info(">>> Restart iteration: " + c);
}
- }, "restart-worker-" + i);
-
- t.start();
+ }
+ catch (Exception e) {
+ err.compareAndSet(null, e);
- threads.add(t);
- }
+ error("Unexpected exception in restart-worker.", e);
+ }
+ }, restartThreads, "restart-worker"));
- for (Thread t : threads)
- t.join(2 * duration);
+ fut.markInitialized();
- for (Thread t : threads) {
- if (t.isAlive())
- t.interrupt();
- }
+ fut.get(getTestTimeout());
if (err.get() != null)
throw err.get();
}
finally {
- stopAllGrids();
+ if (!fut.isDone())
+ fut.cancel();
}
}
@@ -713,147 +696,134 @@ public abstract class
GridCacheAbstractNodeRestartSelfTest extends GridCommonAbs
return;
final long endTime = System.currentTimeMillis() + duration;
-
final AtomicReference<Throwable> err = new AtomicReference<>();
startGrids();
- Collection<Thread> threads = new LinkedList<>();
+ GridCompoundFuture<Long, Long> fut = new GridCompoundFuture<>();
try {
final AtomicInteger txCntr = new AtomicInteger();
-
final CyclicBarrier barrier = new CyclicBarrier(putThreads +
restartThreads);
+ final AtomicInteger startedThreadCntr = new AtomicInteger();
- for (int i = 0; i < putThreads; i++) {
- final int gridIdx = i;
-
- Thread t = new Thread(new Runnable() {
- @Override public void run() {
- try {
- barrier.await();
+ fut.add(GridTestUtils.runMultiThreadedAsync(() -> {
+ try {
+ int gridIdx = startedThreadCntr.getAndIncrement();
- info("Starting put thread: " + gridIdx);
+ barrier.await();
- Ignite ignite = grid(gridIdx);
+ info("Starting put thread: " + gridIdx);
- Thread.currentThread().setName("put-worker-" +
ignite.name());
+ Ignite ignite = grid(gridIdx);
- UUID locNodeId = ignite.cluster().localNode().id();
+ Thread.currentThread().setName("put-worker-" +
ignite.name());
- IgniteCache<Integer, String> cache =
ignite.cache(CACHE_NAME).withAllowAtomicOpsInTx();
+ UUID locNodeId = ignite.cluster().localNode().id();
- List<Integer> keys = new ArrayList<>(txKeys);
+ IgniteCache<Integer, String> cache =
ignite.cache(CACHE_NAME).withAllowAtomicOpsInTx();
- while (System.currentTimeMillis() < endTime &&
err.get() == null) {
- keys.clear();
+ List<Integer> keys = new ArrayList<>(txKeys);
- for (int i = 0; i < txKeys; i++)
- keys.add(RAND.nextInt(keyCnt));
+ while (System.currentTimeMillis() < endTime && err.get()
== null) {
+ keys.clear();
- // Ensure lock order.
- Collections.sort(keys);
+ for (int i = 0; i < txKeys; i++)
+ keys.add(RAND.nextInt(keyCnt));
- int c = 0;
+ // Ensure lock order.
+ Collections.sort(keys);
- try {
- IgniteTransactions txs =
ignite.transactions();
+ int c = 0;
- try (Transaction tx =
txs.txStart(txConcurrency(), REPEATABLE_READ)) {
- c = txCntr.incrementAndGet();
-
- if (c % LOG_FREQ == 0) {
- info(">>> Tx iteration started
[cnt=" + c +
- ", keys=" + keys +
- ", locNodeId=" + locNodeId +
']');
- }
-
- for (int key : keys) {
- int op = cacheOp();
+ try {
+ IgniteTransactions txs = ignite.transactions();
- if (op == 1)
- cache.put(key,
Integer.toString(key));
- else if (op == 2)
- cache.remove(key);
- else
- cache.get(key);
- }
-
- tx.commit();
- }
- }
- catch (IgniteException | CacheException
ignored) {
- // It is ok if primary node leaves grid.
- }
+ try (Transaction tx = txs.txStart(txConcurrency(),
REPEATABLE_READ)) {
+ c = txCntr.incrementAndGet();
if (c % LOG_FREQ == 0) {
- info(">>> Tx iteration finished [cnt=" + c
+
- ", cacheSize=" + cache.localSize() +
+ info(">>> Tx iteration started [cnt=" + c +
", keys=" + keys +
", locNodeId=" + locNodeId + ']');
}
- }
- info(">>> " + Thread.currentThread().getName() + "
finished.");
+ for (int key : keys) {
+ int op = cacheOp();
+
+ if (op == 1)
+ cache.put(key, Integer.toString(key));
+ else if (op == 2)
+ cache.remove(key);
+ else
+ cache.get(key);
+ }
+
+ tx.commit();
+ }
+ }
+ catch (IgniteException | CacheException ignored) {
+ // It is ok if primary node leaves grid.
}
- catch (Exception e) {
- err.compareAndSet(null, e);
- error("Unexpected exception in put-worker.", e);
+ if (c % LOG_FREQ == 0) {
+ info(">>> Tx iteration finished [cnt=" + c +
+ ", cacheSize=" + cache.localSize() +
+ ", keys=" + keys +
+ ", locNodeId=" + locNodeId + ']');
}
}
- }, "put-worker-" + i);
- t.start();
+ info(">>> " + Thread.currentThread().getName() + "
finished.");
+ }
+ catch (Exception e) {
+ err.compareAndSet(null, e);
- threads.add(t);
- }
+ error("Unexpected exception in put-worker.", e);
+ }
+ }, putThreads, "put-worker"));
- for (int i = 0; i < restartThreads; i++) {
- final int gridIdx = i + putThreads;
+ GridTestUtils.waitForCondition(() -> startedThreadCntr.get() ==
putThreads, getTestTimeout());
- Thread t = new Thread(new Runnable() {
- @Override public void run() {
- try {
- barrier.await();
+ fut.add(GridTestUtils.runMultiThreadedAsync(() -> {
+ try {
+ int gridIdx = startedThreadCntr.incrementAndGet();
- info("Starting restart thread: " + gridIdx);
+ barrier.await();
- int cnt = 0;
+ info("Starting restart thread: " + gridIdx);
- while (System.currentTimeMillis() < endTime &&
err.get() == null) {
- stopGrid(getTestIgniteInstanceName(gridIdx),
false, false);
- startGrid(gridIdx);
+ int cnt = 0;
- int c = ++cnt;
+ while (System.currentTimeMillis() < endTime && err.get()
== null) {
+ stopGrid(getTestIgniteInstanceName(gridIdx), false,
false);
+ startGrid(gridIdx);
- if (c % LOG_FREQ == 0)
- info(">>> Restart iteration: " + c);
- }
+ int c = ++cnt;
- info(">>> " + Thread.currentThread().getName() + "
finished.");
- }
- catch (Exception e) {
- err.compareAndSet(null, e);
-
- error("Unexpected exception in restart-worker.",
e);
- }
+ if (c % LOG_FREQ == 0)
+ info(">>> Restart iteration: " + c);
}
- }, "restart-worker-" + i);
- t.start();
+ info(">>> " + Thread.currentThread().getName() + "
finished.");
+ }
+ catch (Exception e) {
+ err.compareAndSet(null, e);
- threads.add(t);
- }
+ error("Unexpected exception in restart-worker.", e);
+ }
+ }, restartThreads, "restart-worker"));
- for (Thread t : threads)
- t.join();
+ fut.markInitialized();
+
+ fut.get(getTestTimeout());
if (err.get() != null)
throw err.get();
}
finally {
- stopAllGrids();
+ if (!fut.isDone())
+ fut.cancel();
}
}
@@ -873,129 +843,116 @@ public abstract class
GridCacheAbstractNodeRestartSelfTest extends GridCommonAbs
startGrids();
- Collection<Thread> threads = new LinkedList<>();
+ GridCompoundFuture<Long, Long> fut = new GridCompoundFuture<>();
try {
final AtomicInteger txCntr = new AtomicInteger();
-
final CyclicBarrier barrier = new CyclicBarrier(putThreads +
restartThreads);
-
+ final AtomicInteger startedThreadCntr = new AtomicInteger();
final int txKeys = 3;
- for (int i = 0; i < putThreads; i++) {
- final int gridIdx = i;
+ fut.add(GridTestUtils.runMultiThreadedAsync(() -> {
+ try {
+ int gridIdx = startedThreadCntr.getAndIncrement();
- Thread t = new Thread(new Runnable() {
- @Override public void run() {
- try {
- barrier.await();
-
- info("Starting put thread: " + gridIdx);
+ barrier.await();
- Ignite ignite = grid(gridIdx);
+ info("Starting put thread: " + gridIdx);
- Thread.currentThread().setName("put-worker-" +
ignite.name());
+ Ignite ignite = grid(gridIdx);
- UUID locNodeId = ignite.cluster().localNode().id();
+ Thread.currentThread().setName("put-worker-" +
ignite.name());
- IgniteCache<Integer, String> cache =
ignite.cache(CACHE_NAME).withAllowAtomicOpsInTx();
+ UUID locNodeId = ignite.cluster().localNode().id();
- List<Integer> keys = new ArrayList<>(txKeys);
+ IgniteCache<Integer, String> cache =
ignite.cache(CACHE_NAME).withAllowAtomicOpsInTx();
- while (System.currentTimeMillis() < endTime &&
err.get() == null) {
- keys.clear();
+ List<Integer> keys = new ArrayList<>(txKeys);
- for (int i = 0; i < txKeys; i++)
- keys.add(RAND.nextInt(keyCnt));
+ while (System.currentTimeMillis() < endTime && err.get()
== null) {
+ keys.clear();
- // Ensure lock order.
- Collections.sort(keys);
+ for (int i = 0; i < txKeys; i++)
+ keys.add(RAND.nextInt(keyCnt));
- int c = 0;
+ // Ensure lock order.
+ Collections.sort(keys);
- try (Transaction tx =
ignite.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
- c = txCntr.incrementAndGet();
+ int c = 0;
- if (c % LOG_FREQ == 0)
- info(">>> Tx iteration started [cnt="
+ c + ", keys=" + keys + ", " +
- "locNodeId=" + locNodeId + ']');
+ try (Transaction tx =
ignite.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ c = txCntr.incrementAndGet();
- Map<Integer, String> batch = new
LinkedHashMap<>();
+ if (c % LOG_FREQ == 0)
+ info(">>> Tx iteration started [cnt=" + c + ",
keys=" + keys + ", " +
+ "locNodeId=" + locNodeId + ']');
- for (int key : keys)
- batch.put(key, String.valueOf(key));
+ Map<Integer, String> batch = new LinkedHashMap<>();
- cache.putAll(batch);
+ for (int key : keys)
+ batch.put(key, String.valueOf(key));
- tx.commit();
- }
- catch (IgniteException | CacheException
ignored) {
- // It is ok if primary node leaves grid.
- }
+ cache.putAll(batch);
- if (c % LOG_FREQ == 0) {
- info(">>> Tx iteration finished [cnt=" + c
+
- ", keys=" + keys + ", " +
- "locNodeId=" + locNodeId + ']');
- }
- }
+ tx.commit();
+ }
+ catch (IgniteException | CacheException ignored) {
+ // It is ok if primary node leaves grid.
}
- catch (Exception e) {
- err.compareAndSet(null, e);
- error("Unexpected exception in put-worker.", e);
+ if (c % LOG_FREQ == 0) {
+ info(">>> Tx iteration finished [cnt=" + c +
+ ", keys=" + keys + ", " +
+ "locNodeId=" + locNodeId + ']');
}
}
- }, "put-worker-" + i);
+ }
+ catch (Exception e) {
+ err.compareAndSet(null, e);
- t.start();
+ error("Unexpected exception in put-worker.", e);
+ }
+ }, putThreads, "put-worker"));
- threads.add(t);
- }
+ GridTestUtils.waitForCondition(() -> startedThreadCntr.get() ==
putThreads, getTestTimeout());
- for (int i = 0; i < restartThreads; i++) {
- final int gridIdx = i + putThreads;
+ fut.add(GridTestUtils.runMultiThreadedAsync(() -> {
+ try {
+ int gridIdx = startedThreadCntr.getAndIncrement();
- Thread t = new Thread(new Runnable() {
- @Override public void run() {
- try {
- barrier.await();
+ barrier.await();
- info("Starting restart thread: " + gridIdx);
+ info("Starting restart thread: " + gridIdx);
- int cnt = 0;
+ int cnt = 0;
- while (System.currentTimeMillis() < endTime &&
err.get() == null) {
- stopGrid(gridIdx);
- startGrid(gridIdx);
+ while (System.currentTimeMillis() < endTime && err.get()
== null) {
+ stopGrid(gridIdx);
+ startGrid(gridIdx);
- int c = ++cnt;
+ int c = ++cnt;
- if (c % LOG_FREQ == 0)
- info(">>> Restart iteration: " + c);
- }
- }
- catch (Exception e) {
- err.compareAndSet(null, e);
-
- error("Unexpected exception in restart-worker.",
e);
- }
+ if (c % LOG_FREQ == 0)
+ info(">>> Restart iteration: " + c);
}
- }, "restart-worker-" + i);
+ }
+ catch (Exception e) {
+ err.compareAndSet(null, e);
- t.start();
+ error("Unexpected exception in restart-worker.", e);
+ }
+ }, restartThreads, "restart-worker"));
- threads.add(t);
- }
+ fut.markInitialized();
- for (Thread t : threads)
- t.join();
+ fut.get(getTestTimeout());
if (err.get() != null)
throw err.get();
}
finally {
- stopAllGrids();
+ if (!fut.isDone())
+ fut.cancel();
}
}
