This is an automated email from the ASF dual-hosted git repository.
DomGarguilo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/main by this push:
new a5979677bc Properly close thread pools (#6480)
a5979677bc is described below
commit a5979677bc3fb9e6eff41623465e4e3bd19cb4c1
Author: Dom G. <[email protected]>
AuthorDate: Mon Jul 20 10:24:35 2026 -0400
Properly close thread pools (#6480)
* close thread pools that were not being closed and also move existing/new
thread pool shutdown calls into a finally block
---
.../core/clientImpl/ClientTabletCacheImplTest.java | 190 +++++-----
.../apache/accumulo/core/crypto/CryptoTest.java | 58 +--
.../core/file/rfile/bcfile/CompressionTest.java | 161 ++++-----
.../core/util/CompletableFutureUtilTest.java | 72 ++--
.../server/tablets/UniqueNameAllocatorTest.java | 56 +--
.../compaction/queue/CompactionJobQueuesTest.java | 86 ++---
.../accumulo/tserver/AssignmentWatcherTest.java | 29 +-
.../accumulo/test/CountNameNodeOpsBulkIT.java | 118 +++---
.../java/org/apache/accumulo/test/LargeReadIT.java | 87 ++---
.../accumulo/test/MultipleManagerFateIT.java | 4 +-
.../org/apache/accumulo/test/ScanServerIT.java | 5 +-
.../accumulo/test/ScanServerMaxLatencyIT.java | 6 +-
.../accumulo/test/TableConfigurationUpdateIT.java | 13 +-
.../apache/accumulo/test/TableOperationsIT.java | 94 ++---
.../accumulo/test/UniqueNameAllocatorIT.java | 95 ++---
.../org/apache/accumulo/test/ZombieScanIT.java | 8 +-
.../test/conf/PropStoreConfigIT_SimpleSuite.java | 277 +++++++-------
.../accumulo/test/fate/meta/ZooMutatorIT.java | 8 +-
.../test/functional/AddSplitIT_SimpleSuite.java | 62 ++--
.../apache/accumulo/test/functional/BulkNewIT.java | 112 +++---
.../accumulo/test/functional/CompactionIT.java | 122 ++++---
.../test/functional/ConcurrentDeleteTableIT.java | 23 +-
.../ConcurrentTableNameOperationsIT.java | 400 +++++++++++----------
.../accumulo/test/functional/FateStarvationIT.java | 54 +--
.../test/functional/FunctionalTestUtils.java | 41 ++-
.../test/functional/ManagerAssignmentIT.java | 72 ++--
.../apache/accumulo/test/functional/ScanIdIT.java | 14 +-
.../apache/accumulo/test/functional/SplitIT.java | 109 +++---
.../accumulo/test/functional/WriteLotsIT.java | 31 +-
29 files changed, 1252 insertions(+), 1155 deletions(-)
diff --git
a/core/src/test/java/org/apache/accumulo/core/clientImpl/ClientTabletCacheImplTest.java
b/core/src/test/java/org/apache/accumulo/core/clientImpl/ClientTabletCacheImplTest.java
index a2cf16c176..9b1cb1adec 100644
---
a/core/src/test/java/org/apache/accumulo/core/clientImpl/ClientTabletCacheImplTest.java
+++
b/core/src/test/java/org/apache/accumulo/core/clientImpl/ClientTabletCacheImplTest.java
@@ -1984,69 +1984,73 @@ public class ClientTabletCacheImplTest {
setLocation(tservers, "tserver3", mte2, ke3, "tserver9");
var executor = Executors.newCachedThreadPool();
- final int lookupCount = 128;
- final int roundCount = 8;
- final int numTasks = roundCount * lookupCount;
-
- List<Future<CachedTablet>> futures = new ArrayList<>(numTasks);
- CountDownLatch startLatch = new CountDownLatch(32); // start a portion of
threads at once
- assertTrue(numTasks >= startLatch.getCount(),
- "Not enough tasks to satisfy latch count - deadlock risk");
-
- // multiple rounds to increase the chance of contention
- for (int round = 0; round < roundCount; round++) {
+ try {
+ final int lookupCount = 128;
+ final int roundCount = 8;
+ final int numTasks = roundCount * lookupCount;
+
+ List<Future<CachedTablet>> futures = new ArrayList<>(numTasks);
+ CountDownLatch startLatch = new CountDownLatch(32); // start a portion
of threads at once
+ assertTrue(numTasks >= startLatch.getCount(),
+ "Not enough tasks to satisfy latch count - deadlock risk");
+
+ // multiple rounds to increase the chance of contention
+ for (int round = 0; round < roundCount; round++) {
+
+ // start a bunch of threads all trying to lookup data in the cache
+ // should see exactly 2 threads doing metadata lookups at a time
+ List<String> rowsToLookup = new ArrayList<>(lookupCount);
+
+ for (int i = 0; i < lookupCount; i++) {
+ String lookup = (char) ('a' + (i % 26)) + "";
+ rowsToLookup.add(lookup);
+ }
- // start a bunch of threads all trying to lookup data in the cache
- // should see exactly 2 threads doing metadata lookups at a time
- List<String> rowsToLookup = new ArrayList<>(lookupCount);
+ Collections.shuffle(rowsToLookup, RANDOM.get());
- for (int i = 0; i < lookupCount; i++) {
- String lookup = (char) ('a' + (i % 26)) + "";
- rowsToLookup.add(lookup);
+ for (var lookup : rowsToLookup) {
+ var future = executor.submit(() -> {
+ startLatch.countDown();
+ startLatch.await();
+ if (RANDOM.get().nextInt(10) < 3) {
+ Thread.yield();
+ }
+ var loc = metaCache.findTablet(context, new Text(lookup), false,
LocationNeed.REQUIRED);
+ if (lookup.compareTo("m") <= 0) {
+ assertEquals("tserver7", loc.getTserverLocation().orElseThrow());
+ } else if (lookup.compareTo("q") <= 0) {
+ assertEquals("tserver8", loc.getTserverLocation().orElseThrow());
+ } else {
+ assertEquals("tserver9", loc.getTserverLocation().orElseThrow());
+ }
+ return loc;
+ });
+ futures.add(future);
+ }
}
+ assertEquals(numTasks, futures.size());
- Collections.shuffle(rowsToLookup, RANDOM.get());
-
- for (var lookup : rowsToLookup) {
- var future = executor.submit(() -> {
- startLatch.countDown();
- startLatch.await();
- if (RANDOM.get().nextInt(10) < 3) {
- Thread.yield();
- }
- var loc = metaCache.findTablet(context, new Text(lookup), false,
LocationNeed.REQUIRED);
- if (lookup.compareTo("m") <= 0) {
- assertEquals("tserver7", loc.getTserverLocation().orElseThrow());
- } else if (lookup.compareTo("q") <= 0) {
- assertEquals("tserver8", loc.getTserverLocation().orElseThrow());
- } else {
- assertEquals("tserver9", loc.getTserverLocation().orElseThrow());
- }
- return loc;
- });
- futures.add(future);
+ for (var future : futures) {
+ assertNotNull(future.get());
}
- }
- assertEquals(numTasks, futures.size());
- for (var future : futures) {
- assertNotNull(future.get());
+ assertTrue(sawTwoActive.get(), "Expected to see exactly two lookups.");
+ // The second metadata tablet (mte2) contains two user tablets (ke2 and
ke3). Depending on
+ // which of these two user tablets is looked up in the metadata table
first will see a total
+ // of 2 or 3 lookups. If the location of ke2 is looked up first then it
will get the locations
+ // of ke2 and ke3 from mte2 and put them in the cache. If the location
of ke3 is looked up
+ // first then it will only get the location of ke3 from mte2 and not ke2.
+ assertTrue(lookups.size() == 2 || lookups.size() == 3,
+ "Expected 2 or 3 lookups, got " + lookups.size() + " : " + lookups);
+ assertEquals(1,
+ lookups.stream().filter(metadataExtent ->
metadataExtent.equals(mte1)).count(),
+ lookups::toString);
+ var mte2Lookups =
+ lookups.stream().filter(metadataExtent ->
metadataExtent.equals(mte2)).count();
+ assertTrue(mte2Lookups == 1 || mte2Lookups == 2, lookups::toString);
+ } finally {
+ executor.shutdown();
}
-
- assertTrue(sawTwoActive.get(), "Expected to see exactly two lookups.");
- // The second metadata tablet (mte2) contains two user tablets (ke2 and
ke3). Depending on which
- // of these two user tablets is looked up in the metadata table first will
see a total of 2 or 3
- // lookups. If the location of ke2 is looked up first then it will get the
locations of ke2 and
- // ke3 from mte2 and put them in the cache. If the location of ke3 is
looked up first then it
- // will only get the location of ke3 from mte2 and not ke2.
- assertTrue(lookups.size() == 2 || lookups.size() == 3,
- "Expected 2 or 3 lookups, got " + lookups.size() + " : " + lookups);
- assertEquals(1, lookups.stream().filter(metadataExtent ->
metadataExtent.equals(mte1)).count(),
- lookups::toString);
- var mte2Lookups =
- lookups.stream().filter(metadataExtent ->
metadataExtent.equals(mte2)).count();
- assertTrue(mte2Lookups == 1 || mte2Lookups == 2, lookups::toString);
- executor.shutdown();
}
@Test
@@ -2097,51 +2101,55 @@ public class ClientTabletCacheImplTest {
metaCache.invalidateCache(ke2);
var executor = Executors.newCachedThreadPool();
+ try {
- // acquire this lock to simulate a metadata table lookup blocking
- lookupLock.lock();
+ // acquire this lock to simulate a metadata table lookup blocking
+ lookupLock.lock();
- // verify test assumption
- assertEquals(0, lookupLock.getQueueLength());
+ // verify test assumption
+ assertEquals(0, lookupLock.getQueueLength());
- // start a background task that will read from the metadata table
- var lookupFuture = executor
- .submit(() -> metaCache.findTablet(context, new Text("h"), false,
LocationNeed.REQUIRED)
- .getTserverLocation().orElseThrow());
+ // start a background task that will read from the metadata table
+ var lookupFuture = executor
+ .submit(() -> metaCache.findTablet(context, new Text("h"), false,
LocationNeed.REQUIRED)
+ .getTserverLocation().orElseThrow());
- // wait for the background thread to get stuck waiting on the lock
- while (lookupLock.getQueueLength() == 0) {
- Thread.sleep(5);
- }
+ // wait for the background thread to get stuck waiting on the lock
+ while (lookupLock.getQueueLength() == 0) {
+ Thread.sleep(5);
+ }
- // The lookup task should be blocked trying to get location from the
metadata table
- assertFalse(lookupFuture.isDone());
- assertEquals(1, lookupLock.getQueueLength());
+ // The lookup task should be blocked trying to get location from the
metadata table
+ assertFalse(lookupFuture.isDone());
+ assertEquals(1, lookupLock.getQueueLength());
- // should be able to get the tablet locations that are still in the cache
w/o blocking
- assertEquals("tserver3",
- metaCache.findTablet(context, new Text("a"), false,
LocationNeed.REQUIRED)
- .getTserverLocation().orElseThrow());
- assertEquals("tserver5",
- metaCache.findTablet(context, new Text("n"), false,
LocationNeed.REQUIRED)
- .getTserverLocation().orElseThrow());
- assertEquals("tserver6",
- metaCache.findTablet(context, new Text("s"), false,
LocationNeed.REQUIRED)
- .getTserverLocation().orElseThrow());
+ // should be able to get the tablet locations that are still in the
cache w/o blocking
+ assertEquals("tserver3",
+ metaCache.findTablet(context, new Text("a"), false,
LocationNeed.REQUIRED)
+ .getTserverLocation().orElseThrow());
+ assertEquals("tserver5",
+ metaCache.findTablet(context, new Text("n"), false,
LocationNeed.REQUIRED)
+ .getTserverLocation().orElseThrow());
+ assertEquals("tserver6",
+ metaCache.findTablet(context, new Text("s"), false,
LocationNeed.REQUIRED)
+ .getTserverLocation().orElseThrow());
- // The lookup task should still be blocked
- assertFalse(lookupFuture.isDone());
- assertEquals(1, lookupLock.getQueueLength());
+ // The lookup task should still be blocked
+ assertFalse(lookupFuture.isDone());
+ assertEquals(1, lookupLock.getQueueLength());
- // allow the metadata lookup running in background thread to proceed.
- lookupLock.unlock();
+ // allow the metadata lookup running in background thread to proceed.
+ lookupLock.unlock();
- // The future should be able to run and complete
- assertEquals("tserver4", lookupFuture.get());
+ // The future should be able to run and complete
+ assertEquals("tserver4", lookupFuture.get());
- // verify test assumptions
- assertTrue(lookupFuture.isDone());
- assertEquals(0, lookupLock.getQueueLength());
+ // verify test assumptions
+ assertTrue(lookupFuture.isDone());
+ assertEquals(0, lookupLock.getQueueLength());
+ } finally {
+ executor.shutdown();
+ }
}
@Test
diff --git a/core/src/test/java/org/apache/accumulo/core/crypto/CryptoTest.java
b/core/src/test/java/org/apache/accumulo/core/crypto/CryptoTest.java
index b7833bedeb..668df2a26b 100644
--- a/core/src/test/java/org/apache/accumulo/core/crypto/CryptoTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/crypto/CryptoTest.java
@@ -560,34 +560,38 @@ public class CryptoTest {
byte[] cipherText = out.toByteArray();
var executor = Executors.newCachedThreadPool();
+ try {
+
+ final int numTasks = 32;
+ List<Future<Boolean>> verifyFutures = new ArrayList<>(numTasks);
+ CountDownLatch startLatch = new CountDownLatch(numTasks);
+ assertTrue(numTasks >= startLatch.getCount(),
+ "Not enough tasks to satisfy latch count - deadlock risk");
+
+ FileDecrypter decrypter = cs.getFileDecrypter(new
CryptoEnvironmentImpl(scope, null, params));
+
+ // verify that each input stream returned by decrypter.decryptStream()
is independent when
+ // used by multiple threads
+ for (int i = 0; i < numTasks; i++) {
+ var future = executor.submit(() -> {
+ startLatch.countDown();
+ startLatch.await();
+ try (ByteArrayInputStream in = new ByteArrayInputStream(cipherText);
+ DataInputStream decrypted = new
DataInputStream(decrypter.decryptStream(in))) {
+ byte[] dataRead = new byte[plainText.length];
+ decrypted.readFully(dataRead);
+ return Arrays.equals(plainText, dataRead);
+ }
+ });
+ verifyFutures.add(future);
+ }
+ assertEquals(numTasks, verifyFutures.size());
- final int numTasks = 32;
- List<Future<Boolean>> verifyFutures = new ArrayList<>(numTasks);
- CountDownLatch startLatch = new CountDownLatch(numTasks);
- assertTrue(numTasks >= startLatch.getCount(),
- "Not enough tasks to satisfy latch count - deadlock risk");
-
- FileDecrypter decrypter = cs.getFileDecrypter(new
CryptoEnvironmentImpl(scope, null, params));
-
- // verify that each input stream returned by decrypter.decryptStream() is
independent when used
- // by multiple threads
- for (int i = 0; i < numTasks; i++) {
- var future = executor.submit(() -> {
- startLatch.countDown();
- startLatch.await();
- try (ByteArrayInputStream in = new ByteArrayInputStream(cipherText);
- DataInputStream decrypted = new
DataInputStream(decrypter.decryptStream(in))) {
- byte[] dataRead = new byte[plainText.length];
- decrypted.readFully(dataRead);
- return Arrays.equals(plainText, dataRead);
- }
- });
- verifyFutures.add(future);
- }
- assertEquals(numTasks, verifyFutures.size());
-
- for (var future : verifyFutures) {
- assertTrue(future.get());
+ for (var future : verifyFutures) {
+ assertTrue(future.get());
+ }
+ } finally {
+ executor.shutdownNow();
}
}
diff --git
a/core/src/test/java/org/apache/accumulo/core/file/rfile/bcfile/CompressionTest.java
b/core/src/test/java/org/apache/accumulo/core/file/rfile/bcfile/CompressionTest.java
index ccbb1f0974..c4c2f9ca30 100644
---
a/core/src/test/java/org/apache/accumulo/core/file/rfile/bcfile/CompressionTest.java
+++
b/core/src/test/java/org/apache/accumulo/core/file/rfile/bcfile/CompressionTest.java
@@ -18,7 +18,6 @@
*/
package org.apache.accumulo.core.file.rfile.bcfile;
-import static java.util.concurrent.TimeUnit.SECONDS;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -198,33 +197,31 @@ public class CompressionTest {
final int numTasks = 32;
ExecutorService service = Executors.newFixedThreadPool(numTasks);
-
- ArrayList<Future<Boolean>> results = new ArrayList<>(numTasks);
- CountDownLatch startLatch = new CountDownLatch(numTasks);
- assertTrue(numTasks >= startLatch.getCount(),
- "Not enough tasks/threads to satisfy latch count - deadlock risk");
-
- for (int i = 0; i < numTasks; i++) {
- results.add(service.submit(() -> {
- startLatch.countDown();
- startLatch.await();
- assertNotNull(al.getCodec(), al + " should not be null");
- return true;
- }));
- }
- assertEquals(numTasks, results.size());
-
- service.shutdown();
-
- assertNotNull(codec, al + " should not be null");
-
- while (!service.awaitTermination(1, SECONDS)) {
- // wait
- }
-
- for (Future<Boolean> result : results) {
- assertTrue(result.get(),
- al + " resulted in a failed call to getcodec within the thread
pool");
+ try {
+
+ ArrayList<Future<Boolean>> results = new ArrayList<>(numTasks);
+ CountDownLatch startLatch = new CountDownLatch(numTasks);
+ assertTrue(numTasks >= startLatch.getCount(),
+ "Not enough tasks/threads to satisfy latch count - deadlock
risk");
+
+ for (int i = 0; i < numTasks; i++) {
+ results.add(service.submit(() -> {
+ startLatch.countDown();
+ startLatch.await();
+ assertNotNull(al.getCodec(), al + " should not be null");
+ return true;
+ }));
+ }
+ assertEquals(numTasks, results.size());
+
+ assertNotNull(codec, al + " should not be null");
+
+ for (Future<Boolean> result : results) {
+ assertTrue(result.get(),
+ al + " resulted in a failed call to getcodec within the thread
pool");
+ }
+ } finally {
+ service.shutdown();
}
somethingGotTested = true;
}
@@ -247,32 +244,30 @@ public class CompressionTest {
final int numTasks = 32;
ExecutorService service = Executors.newFixedThreadPool(numTasks);
-
- ArrayList<Future<Boolean>> results = new ArrayList<>(numTasks);
- CountDownLatch startLatch = new CountDownLatch(numTasks);
- assertTrue(numTasks >= startLatch.getCount(),
- "Not enough tasks/threads to satisfy latch count - deadlock risk");
-
- for (int i = 0; i < numTasks; i++) {
-
- results.add(service.submit(() -> {
- startLatch.countDown();
- startLatch.await();
- assertNotNull(al.getCodec(), al + " should have a non-null codec");
- return true;
- }));
- }
- assertEquals(numTasks, results.size());
-
- service.shutdown();
-
- while (!service.awaitTermination(1, SECONDS)) {
- // wait
- }
-
- for (Future<Boolean> result : results) {
- assertTrue(result.get(),
- al + " resulted in a failed call to getcodec within the thread
pool");
+ try {
+
+ ArrayList<Future<Boolean>> results = new ArrayList<>(numTasks);
+ CountDownLatch startLatch = new CountDownLatch(numTasks);
+ assertTrue(numTasks >= startLatch.getCount(),
+ "Not enough tasks/threads to satisfy latch count - deadlock
risk");
+
+ for (int i = 0; i < numTasks; i++) {
+
+ results.add(service.submit(() -> {
+ startLatch.countDown();
+ startLatch.await();
+ assertNotNull(al.getCodec(), al + " should have a non-null
codec");
+ return true;
+ }));
+ }
+ assertEquals(numTasks, results.size());
+
+ for (Future<Boolean> result : results) {
+ assertTrue(result.get(),
+ al + " resulted in a failed call to getcodec within the thread
pool");
+ }
+ } finally {
+ service.shutdown();
}
somethingGotTested = true;
}
@@ -295,37 +290,37 @@ public class CompressionTest {
final int numTasks = 32;
ExecutorService service = Executors.newFixedThreadPool(numTasks);
-
- ArrayList<Callable<Integer>> list = new ArrayList<>(numTasks);
-
- CountDownLatch startLatch = new CountDownLatch(numTasks);
- assertTrue(numTasks >= startLatch.getCount(),
- "Not enough tasks/threads to satisfy latch count - deadlock risk");
-
- // keep track of the system's identity hashcodes.
-
- for (int i = 0; i < numTasks; i++) {
- list.add(() -> {
- startLatch.countDown();
- startLatch.await();
- CompressionCodec codec = al.getCodec();
- assertNotNull(codec, al + " resulted in a non-null codec");
- return System.identityHashCode(codec);
- });
- }
- assertEquals(numTasks, list.size());
-
- final HashSet<Integer> hashCodes = new HashSet<>();
- for (Future<Integer> result : service.invokeAll(list)) {
- hashCodes.add(result.get());
+ try {
+
+ ArrayList<Callable<Integer>> list = new ArrayList<>(numTasks);
+
+ CountDownLatch startLatch = new CountDownLatch(numTasks);
+ assertTrue(numTasks >= startLatch.getCount(),
+ "Not enough tasks/threads to satisfy latch count - deadlock
risk");
+
+ // keep track of the system's identity hashcodes.
+
+ for (int i = 0; i < numTasks; i++) {
+ list.add(() -> {
+ startLatch.countDown();
+ startLatch.await();
+ CompressionCodec codec = al.getCodec();
+ assertNotNull(codec, al + " resulted in a non-null codec");
+ return System.identityHashCode(codec);
+ });
+ }
+ assertEquals(numTasks, list.size());
+
+ final HashSet<Integer> hashCodes = new HashSet<>();
+ for (Future<Integer> result : service.invokeAll(list)) {
+ hashCodes.add(result.get());
+ }
+ assertEquals(1, hashCodes.size(), al + " created too many codecs");
+
+ } finally {
+ service.shutdown();
}
- assertEquals(1, hashCodes.size(), al + " created too many codecs");
-
- service.shutdown();
- while (!service.awaitTermination(1, SECONDS)) {
- // wait
- }
somethingGotTested = true;
}
}
diff --git
a/core/src/test/java/org/apache/accumulo/core/util/CompletableFutureUtilTest.java
b/core/src/test/java/org/apache/accumulo/core/util/CompletableFutureUtilTest.java
index 2e719540b6..deaa61a014 100644
---
a/core/src/test/java/org/apache/accumulo/core/util/CompletableFutureUtilTest.java
+++
b/core/src/test/java/org/apache/accumulo/core/util/CompletableFutureUtilTest.java
@@ -62,40 +62,44 @@ public class CompletableFutureUtilTest {
@Test
public void testIterateUntil() throws Exception {
ExecutorService es = Executors.newFixedThreadPool(1);
- Function<Integer,CompletableFuture<Integer>> step =
- n -> CompletableFuture.supplyAsync(() -> n - 1, es);
- Predicate<Integer> isDone = n -> n == 0;
- // The call stack should overflow before 10,000 calls, so this
- // effectively tests whether iterateUntil avoids stack overflows
- // when given async futures.
- for (int n : new int[] {0, 1, 2, 3, 100, 10_000}) {
- assertEquals(0, CompletableFutureUtil.iterateUntil(step, isDone,
n).get());
- }
- // Test throwing an exception in the step function.
- {
- Function<Integer,CompletableFuture<Integer>> badStep = n -> {
- throw new RuntimeException();
- };
- assertThrows(ExecutionException.class,
- () -> CompletableFutureUtil.iterateUntil(badStep, isDone,
100).get());
- }
- // Test throwing an exception in the future returned by the step
- // function.
- {
- Function<Integer,CompletableFuture<Integer>> badStep =
- n -> CompletableFuture.supplyAsync(() -> {
- throw new RuntimeException();
- }, es);
- assertThrows(ExecutionException.class,
- () -> CompletableFutureUtil.iterateUntil(badStep, isDone,
100).get());
- }
- // Test throwing an exception in the predicate.
- {
- Predicate<Integer> badIsDone = n -> {
- throw new RuntimeException();
- };
- assertThrows(ExecutionException.class,
- () -> CompletableFutureUtil.iterateUntil(step, badIsDone,
100).get());
+ try {
+ Function<Integer,CompletableFuture<Integer>> step =
+ n -> CompletableFuture.supplyAsync(() -> n - 1, es);
+ Predicate<Integer> isDone = n -> n == 0;
+ // The call stack should overflow before 10,000 calls, so this
+ // effectively tests whether iterateUntil avoids stack overflows
+ // when given async futures.
+ for (int n : new int[] {0, 1, 2, 3, 100, 10_000}) {
+ assertEquals(0, CompletableFutureUtil.iterateUntil(step, isDone,
n).get());
+ }
+ // Test throwing an exception in the step function.
+ {
+ Function<Integer,CompletableFuture<Integer>> badStep = n -> {
+ throw new RuntimeException();
+ };
+ assertThrows(ExecutionException.class,
+ () -> CompletableFutureUtil.iterateUntil(badStep, isDone,
100).get());
+ }
+ // Test throwing an exception in the future returned by the step
+ // function.
+ {
+ Function<Integer,CompletableFuture<Integer>> badStep =
+ n -> CompletableFuture.supplyAsync(() -> {
+ throw new RuntimeException();
+ }, es);
+ assertThrows(ExecutionException.class,
+ () -> CompletableFutureUtil.iterateUntil(badStep, isDone,
100).get());
+ }
+ // Test throwing an exception in the predicate.
+ {
+ Predicate<Integer> badIsDone = n -> {
+ throw new RuntimeException();
+ };
+ assertThrows(ExecutionException.class,
+ () -> CompletableFutureUtil.iterateUntil(step, badIsDone,
100).get());
+ }
+ } finally {
+ es.shutdown();
}
}
}
diff --git
a/server/base/src/test/java/org/apache/accumulo/server/tablets/UniqueNameAllocatorTest.java
b/server/base/src/test/java/org/apache/accumulo/server/tablets/UniqueNameAllocatorTest.java
index 6120cf6310..6707061be8 100644
---
a/server/base/src/test/java/org/apache/accumulo/server/tablets/UniqueNameAllocatorTest.java
+++
b/server/base/src/test/java/org/apache/accumulo/server/tablets/UniqueNameAllocatorTest.java
@@ -64,37 +64,41 @@ public class UniqueNameAllocatorTest {
public void testConcurrent() throws Exception {
NameIterator iter = new NameIterator(0, 10_000);
var executor = Executors.newCachedThreadPool();
+ try {
- // start 100 threads each consuming 100 names
- List<Future<String[]>> futures = new ArrayList<>(100);
- for (int i = 0; i < 100; i++) {
- var future = executor.submit(() -> {
- String[] names = new String[100];
- for (int j = 0; j < 100; j++) {
- assertTrue(iter.hasNext());
- names[j] = iter.next();
- }
- return names;
- });
- futures.add(future);
- }
+ // start 100 threads each consuming 100 names
+ List<Future<String[]>> futures = new ArrayList<>(100);
+ for (int i = 0; i < 100; i++) {
+ var future = executor.submit(() -> {
+ String[] names = new String[100];
+ for (int j = 0; j < 100; j++) {
+ assertTrue(iter.hasNext());
+ names[j] = iter.next();
+ }
+ return names;
+ });
+ futures.add(future);
+ }
- // verify 10_000 unique names were generated
- TreeSet<String> allNames = new TreeSet<>();
- for (var future : futures) {
- String[] names = future.get();
- for (String name : names) {
- assertTrue(allNames.add(name));
+ // verify 10_000 unique names were generated
+ TreeSet<String> allNames = new TreeSet<>();
+ for (var future : futures) {
+ String[] names = future.get();
+ for (String name : names) {
+ assertTrue(allNames.add(name));
+ }
}
- }
- // everything should be consumed from the iterator
- assertFalse(iter.hasNext());
+ // everything should be consumed from the iterator
+ assertFalse(iter.hasNext());
- // verify order of names
- long expected = 0;
- for (String name : allNames) {
- assertEquals(expected++, Long.parseLong(name, 36));
+ // verify order of names
+ long expected = 0;
+ for (String name : allNames) {
+ assertEquals(expected++, Long.parseLong(name, 36));
+ }
+ } finally {
+ executor.shutdown();
}
}
}
diff --git
a/server/manager/src/test/java/org/apache/accumulo/manager/compaction/queue/CompactionJobQueuesTest.java
b/server/manager/src/test/java/org/apache/accumulo/manager/compaction/queue/CompactionJobQueuesTest.java
index d34cc2666d..016a784704 100644
---
a/server/manager/src/test/java/org/apache/accumulo/manager/compaction/queue/CompactionJobQueuesTest.java
+++
b/server/manager/src/test/java/org/apache/accumulo/manager/compaction/queue/CompactionJobQueuesTest.java
@@ -270,55 +270,57 @@ public class CompactionJobQueuesTest {
Stream.of("G1", "G2",
"G3").map(ResourceGroupId::of).toArray(ResourceGroupId[]::new);
var executor = Executors.newFixedThreadPool(groups.length);
+ try {
- List<Future<Integer>> futures = new ArrayList<>();
+ List<Future<Integer>> futures = new ArrayList<>();
- AtomicBoolean stop = new AtomicBoolean(false);
+ AtomicBoolean stop = new AtomicBoolean(false);
- // create a background thread per a group that polls jobs for the group
- for (var group : groups) {
- var future = executor.submit(() -> {
- int seen = 0;
- while (!stop.get()) {
- var job = jobQueues.poll(group);
- if (job != null) {
- seen++;
+ // create a background thread per a group that polls jobs for the group
+ for (var group : groups) {
+ var future = executor.submit(() -> {
+ int seen = 0;
+ while (!stop.get()) {
+ var job = jobQueues.poll(group);
+ if (job != null) {
+ seen++;
+ }
}
- }
-
- // After stop was set, nothing should be added to queues anymore.
Drain anything that is
- // present and then exit.
- while (jobQueues.poll(group) != null) {
- seen++;
- }
-
- return seen;
- });
- futures.add(future);
- }
-
- // Add jobs to queues spread across the groups. While these are being
added the background
- // threads should concurrently empty queues causing them to be deleted.
- for (int i = 0; i < numToAdd; i++) {
- // Create unique exents because re-adding the same extent will clobber
any jobs already in the
- // queue for that extent which could throw off the counts
- KeyExtent extent = new KeyExtent(TableId.of("1"), new Text(i + "z"), new
Text(i + "a"));
- jobQueues.add(extent, List.of(newJob((short) (i % 31), i, groups[i %
groups.length])));
- }
- // Cause the background threads to exit after polling all data
- stop.set(true);
+ // After stop was set, nothing should be added to queues anymore.
Drain anything that is
+ // present and then exit.
+ while (jobQueues.poll(group) != null) {
+ seen++;
+ }
- // Count the total jobs seen by background threads
- int totalSeen = 0;
- for (var future : futures) {
- totalSeen += future.get();
+ return seen;
+ });
+ futures.add(future);
+ }
+
+ // Add jobs to queues spread across the groups. While these are being
added the background
+ // threads should concurrently empty queues causing them to be deleted.
+ for (int i = 0; i < numToAdd; i++) {
+ // Create unique exents because re-adding the same extent will clobber
any jobs already in
+ // the queue for that extent which could throw off the counts
+ KeyExtent extent = new KeyExtent(TableId.of("1"), new Text(i + "z"),
new Text(i + "a"));
+ jobQueues.add(extent, List.of(newJob((short) (i % 31), i, groups[i %
groups.length])));
+ }
+
+ // Cause the background threads to exit after polling all data
+ stop.set(true);
+
+ // Count the total jobs seen by background threads
+ int totalSeen = 0;
+ for (var future : futures) {
+ totalSeen += future.get();
+ }
+
+ // The background threads should have seen every job that was added
+ assertEquals(numToAdd, totalSeen);
+ } finally {
+ executor.shutdown();
}
-
- executor.shutdown();
-
- // The background threads should have seen every job that was added
- assertEquals(numToAdd, totalSeen);
}
@Test
diff --git
a/server/tserver/src/test/java/org/apache/accumulo/tserver/AssignmentWatcherTest.java
b/server/tserver/src/test/java/org/apache/accumulo/tserver/AssignmentWatcherTest.java
index 4a90cca43f..c0bdc1d1d1 100644
---
a/server/tserver/src/test/java/org/apache/accumulo/tserver/AssignmentWatcherTest.java
+++
b/server/tserver/src/test/java/org/apache/accumulo/tserver/AssignmentWatcherTest.java
@@ -40,21 +40,26 @@ public class AssignmentWatcherTest {
@Test
public void testAssignmentWarning() {
var e = (ScheduledThreadPoolExecutor) Executors.newScheduledThreadPool(1);
- var c = new
ConfigurationCopy(Map.of(Property.TSERV_ASSIGNMENT_DURATION_WARNING.getKey(),
"0"));
- ServerContext context = createMock(ServerContext.class);
- expect(context.getScheduledExecutor()).andReturn(e);
- expect(context.getConfiguration()).andReturn(c);
+ try {
+ var c =
+ new
ConfigurationCopy(Map.of(Property.TSERV_ASSIGNMENT_DURATION_WARNING.getKey(),
"0"));
+ ServerContext context = createMock(ServerContext.class);
+ expect(context.getScheduledExecutor()).andReturn(e);
+ expect(context.getConfiguration()).andReturn(c);
- ActiveAssignmentRunnable task = createMock(ActiveAssignmentRunnable.class);
- expect(task.getException()).andReturn(new Exception("Assignment warning
happened"));
+ ActiveAssignmentRunnable task =
createMock(ActiveAssignmentRunnable.class);
+ expect(task.getException()).andReturn(new Exception("Assignment warning
happened"));
- var assignments = Map.of(new KeyExtent(TableId.of("1"), null, null),
- new RunnableStartedAt(task, System.currentTimeMillis()));
- var watcher = new AssignmentWatcher(context, assignments);
+ var assignments = Map.of(new KeyExtent(TableId.of("1"), null, null),
+ new RunnableStartedAt(task, System.currentTimeMillis()));
+ var watcher = new AssignmentWatcher(context, assignments);
- replay(context, task);
- watcher.run();
- verify(context, task);
+ replay(context, task);
+ watcher.run();
+ verify(context, task);
+ } finally {
+ e.shutdownNow();
+ }
}
}
diff --git
a/test/src/main/java/org/apache/accumulo/test/CountNameNodeOpsBulkIT.java
b/test/src/main/java/org/apache/accumulo/test/CountNameNodeOpsBulkIT.java
index 8deb58a6a4..4ca0835868 100644
--- a/test/src/main/java/org/apache/accumulo/test/CountNameNodeOpsBulkIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/CountNameNodeOpsBulkIT.java
@@ -18,7 +18,6 @@
*/
package org.apache.accumulo.test;
-import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.accumulo.core.util.LazySingletons.GSON;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -124,66 +123,69 @@ public class CountNameNodeOpsBulkIT extends
ConfigurableMacBase {
fs.mkdirs(base);
ExecutorService es = Executors.newFixedThreadPool(5);
- List<Future<String>> futures = new ArrayList<>();
- for (int i = 0; i < 10; i++) {
- final int which = i;
- futures.add(es.submit(() -> {
- Path files = new Path(base, "files" + which);
- fs.mkdirs(files);
- for (int i1 = 0; i1 < 100; i1++) {
- FileSKVWriter writer =
FileOperations.getInstance().newWriterBuilder()
- .forFile(
- UnreferencedTabletFile.of(fs,
- new Path(files + "/bulk_" + i1 + "." +
RFile.EXTENSION)),
- fs, fs.getConf(), NoCryptoServiceFactory.NONE)
-
.withTableConfiguration(DefaultConfiguration.getInstance()).build();
- writer.startDefaultLocalityGroup();
- for (int j = 0x100; j < 0xfff; j += 3) {
- writer.append(new Key(Integer.toHexString(j)), new Value());
+ try {
+ List<Future<String>> futures = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ final int which = i;
+ futures.add(es.submit(() -> {
+ Path files = new Path(base, "files" + which);
+ fs.mkdirs(files);
+ for (int i1 = 0; i1 < 100; i1++) {
+ FileSKVWriter writer =
FileOperations.getInstance().newWriterBuilder()
+ .forFile(
+ UnreferencedTabletFile.of(fs,
+ new Path(files + "/bulk_" + i1 + "." +
RFile.EXTENSION)),
+ fs, fs.getConf(), NoCryptoServiceFactory.NONE)
+
.withTableConfiguration(DefaultConfiguration.getInstance()).build();
+ writer.startDefaultLocalityGroup();
+ for (int j = 0x100; j < 0xfff; j += 3) {
+ writer.append(new Key(Integer.toHexString(j)), new Value());
+ }
+ writer.close();
}
- writer.close();
+ return files.toString();
+ }));
+ }
+ List<String> dirs = new ArrayList<>();
+ for (Future<String> f : futures) {
+ dirs.add(f.get());
+ }
+ log.info("Importing");
+ long startOps = getStat(getStats(), "FileInfoOps");
+ long now = System.currentTimeMillis();
+ List<Future<Object>> errs = new ArrayList<>();
+ for (String dir : dirs) {
+ errs.add(es.submit(() -> {
+ c.tableOperations().importDirectory(dir).to(tableName).load();
+ return null;
+ }));
+ }
+ for (Future<Object> err : errs) {
+ err.get();
+ }
+ log.info(
+ String.format("Completed in %.2f seconds",
(System.currentTimeMillis() - now) / 1000.));
+ Thread.sleep(SECONDS.toMillis(30));
+ Map<?,?> map = getStats();
+ map.forEach((k, v) -> {
+ try {
+ if (v != null && Double.parseDouble(v.toString()) > 0.0) {
+ log.debug("{}:{}", k, v);
+ }
+ } catch (NumberFormatException e) {
+ // only looking for numbers
}
- return files.toString();
- }));
- }
- List<String> dirs = new ArrayList<>();
- for (Future<String> f : futures) {
- dirs.add(f.get());
- }
- log.info("Importing");
- long startOps = getStat(getStats(), "FileInfoOps");
- long now = System.currentTimeMillis();
- List<Future<Object>> errs = new ArrayList<>();
- for (String dir : dirs) {
- errs.add(es.submit(() -> {
- c.tableOperations().importDirectory(dir).to(tableName).load();
- return null;
- }));
- }
- for (Future<Object> err : errs) {
- err.get();
+ });
+ long getFileInfoOpts = getStat(map, "FileInfoOps") - startOps;
+ log.info("New bulk import used {} opts, vs old using 2060",
getFileInfoOpts);
+ // counts for old bulk import:
+ // Expected number of FileInfoOps was between 1000 and 2100
+ // new bulk import is way better :)
+ assertEquals(20, getFileInfoOpts, "unexpected number of FileInfoOps");
+
+ } finally {
+ es.shutdown();
}
- es.shutdown();
- es.awaitTermination(2, MINUTES);
- log.info(
- String.format("Completed in %.2f seconds",
(System.currentTimeMillis() - now) / 1000.));
- Thread.sleep(SECONDS.toMillis(30));
- Map<?,?> map = getStats();
- map.forEach((k, v) -> {
- try {
- if (v != null && Double.parseDouble(v.toString()) > 0.0) {
- log.debug("{}:{}", k, v);
- }
- } catch (NumberFormatException e) {
- // only looking for numbers
- }
- });
- long getFileInfoOpts = getStat(map, "FileInfoOps") - startOps;
- log.info("New bulk import used {} opts, vs old using 2060",
getFileInfoOpts);
- // counts for old bulk import:
- // Expected number of FileInfoOps was between 1000 and 2100
- // new bulk import is way better :)
- assertEquals(20, getFileInfoOpts, "unexpected number of FileInfoOps");
}
}
}
diff --git a/test/src/main/java/org/apache/accumulo/test/LargeReadIT.java
b/test/src/main/java/org/apache/accumulo/test/LargeReadIT.java
index b3f0cdc305..750da22d3c 100644
--- a/test/src/main/java/org/apache/accumulo/test/LargeReadIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/LargeReadIT.java
@@ -127,52 +127,57 @@ public class LargeReadIT extends AccumuloClusterHarness {
final int numTasks = 64;
var executor = Executors.newFixedThreadPool(numTasks);
- CountDownLatch startLatch = new CountDownLatch(numTasks);
- assertTrue(numTasks >= startLatch.getCount(),
- "Not enough tasks/threads to satisfy latch count - deadlock risk");
- Callable<Long> scanTask = () -> {
- startLatch.countDown();
- startLatch.await();
- try (var scanner = client.createScanner(tableName)) {
- scannerConfigurer.accept(scanner);
- return scanner.stream().count();
+ try {
+ CountDownLatch startLatch = new CountDownLatch(numTasks);
+ assertTrue(numTasks >= startLatch.getCount(),
+ "Not enough tasks/threads to satisfy latch count - deadlock risk");
+ Callable<Long> scanTask = () -> {
+ startLatch.countDown();
+ startLatch.await();
+ try (var scanner = client.createScanner(tableName)) {
+ scannerConfigurer.accept(scanner);
+ return scanner.stream().count();
+ }
+ };
+
+ // Run lots of concurrent task that should only read the small data,
if they read the big
+ // column family then it will exceed the tablet server memory and
cause it to die and the
+ // test to timeout.
+ var tasks =
+ Stream.iterate(scanTask, t -> t).limit(numTasks *
5).collect(Collectors.toList());
+ assertEquals(numTasks * 5, tasks.size());
+ for (var future : executor.invokeAll(tasks)) {
+ assertEquals(100, future.get());
}
- };
-
- // Run lots of concurrent task that should only read the small data, if
they read the big
- // column family then it will exceed the tablet server memory and cause
it to die and the test
- // to timeout.
- var tasks = Stream.iterate(scanTask, t -> t).limit(numTasks *
5).collect(Collectors.toList());
- assertEquals(numTasks * 5, tasks.size());
- for (var future : executor.invokeAll(tasks)) {
- assertEquals(100, future.get());
- }
- // expected data to be in memory for this part of the test, so verify
that
- var ctx = getCluster().getServerContext();
- try (var tablets =
ctx.getAmple().readTablets().forTable(ctx.getTableId(tableName))
- .fetch(TabletMetadata.ColumnType.FILES).build()) {
- assertEquals(0, tablets.stream().count());
- }
+ // expected data to be in memory for this part of the test, so verify
that
+ var ctx = getCluster().getServerContext();
+ try (var tablets =
ctx.getAmple().readTablets().forTable(ctx.getTableId(tableName))
+ .fetch(TabletMetadata.ColumnType.FILES).build()) {
+ assertEquals(0, tablets.stream().count());
+ }
- // flush data and verify
- client.tableOperations().flush(tableName, null, null, true);
- try (var tablets =
ctx.getAmple().readTablets().forTable(ctx.getTableId(tableName))
- .fetch(TabletMetadata.ColumnType.FILES).build()) {
- assertEquals(1, tablets.stream().count());
- }
+ // flush data and verify
+ client.tableOperations().flush(tableName, null, null, true);
+ try (var tablets =
ctx.getAmple().readTablets().forTable(ctx.getTableId(tableName))
+ .fetch(TabletMetadata.ColumnType.FILES).build()) {
+ assertEquals(1, tablets.stream().count());
+ }
- // Run the scans again, reading from files instead of in memory map...
verify the large data
- // is not brought into memory from the file, which would kill the tablet
server. The test was
- // created because of a bug in the native map code, but can also check
the rfile code for a
- // similar problem.
- for (var future : executor.invokeAll(tasks)) {
- assertEquals(100, future.get());
- }
+ // Run the scans again, reading from files instead of in memory map...
verify the large data
+ // is not brought into memory from the file, which would kill the
tablet server. The test
+ // was created because of a bug in the native map code, but can also
check the rfile code
+ // for a similar problem.
+ for (var future : executor.invokeAll(tasks)) {
+ assertEquals(100, future.get());
+ }
- // this test assumes the first key in the tablet is the big one, verify
this assumption
- try (var scanner = client.createScanner(tableName)) {
- assertEquals(10_000_000,
scanner.iterator().next().getValue().getSize());
+ // this test assumes the first key in the tablet is the big one,
verify this assumption
+ try (var scanner = client.createScanner(tableName)) {
+ assertEquals(10_000_000,
scanner.iterator().next().getValue().getSize());
+ }
+ } finally {
+ executor.shutdown();
}
}
}
diff --git
a/test/src/main/java/org/apache/accumulo/test/MultipleManagerFateIT.java
b/test/src/main/java/org/apache/accumulo/test/MultipleManagerFateIT.java
index dce79c9953..a15b1777f6 100644
--- a/test/src/main/java/org/apache/accumulo/test/MultipleManagerFateIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/MultipleManagerFateIT.java
@@ -245,10 +245,10 @@ public class MultipleManagerFateIT extends
ConfigurableMacBase {
// Check that each background thread made a least one loop over all
its table operations.
assertTrue(loops > 0);
}
+ } finally {
+ executor.shutdown();
}
- executor.shutdown();
-
}
private static void waitToSeeManagers(ClientContext context, int
expectedManagers,
diff --git a/test/src/main/java/org/apache/accumulo/test/ScanServerIT.java
b/test/src/main/java/org/apache/accumulo/test/ScanServerIT.java
index 0c3140c38d..e82299c576 100644
--- a/test/src/main/java/org/apache/accumulo/test/ScanServerIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/ScanServerIT.java
@@ -320,8 +320,9 @@ public class ScanServerIT extends SharedMiniClusterBase {
var e = assertThrows(ExecutionException.class, () -> f.get());
assertTrue(e.getCause() instanceof AssertionError);
});
- } // when the scanner is closed, all open sessions should be closed
- executor.shutdown();
+ } finally { // when the scanner is closed, all open sessions should be
closed
+ executor.shutdown();
+ }
}
}
diff --git
a/test/src/main/java/org/apache/accumulo/test/ScanServerMaxLatencyIT.java
b/test/src/main/java/org/apache/accumulo/test/ScanServerMaxLatencyIT.java
index 62d67a99fa..b92556991b 100644
--- a/test/src/main/java/org/apache/accumulo/test/ScanServerMaxLatencyIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/ScanServerMaxLatencyIT.java
@@ -150,14 +150,14 @@ public class ScanServerMaxLatencyIT extends
ConfigurableMacBase {
// The write task should still be running unless they experienced an
exception.
assertTrue(futures.stream().noneMatch(Future::isDone));
- executor.shutdownNow();
- executor.awaitTermination(600, TimeUnit.SECONDS);
-
assertEquals(-1, readMaxElapsed(client, EVENTUAL, table2));
// Now that nothing is writing its expected that max read by an
immediate scan will see any
// data an eventual scan would see.
assertTrue(
readMaxElapsed(client, IMMEDIATE, table1) >= readMaxElapsed(client,
EVENTUAL, table1));
+ } finally {
+ executor.shutdownNow();
+ executor.awaitTermination(600, TimeUnit.SECONDS);
}
}
diff --git
a/test/src/main/java/org/apache/accumulo/test/TableConfigurationUpdateIT.java
b/test/src/main/java/org/apache/accumulo/test/TableConfigurationUpdateIT.java
index 27899f38d9..23fd5a4908 100644
---
a/test/src/main/java/org/apache/accumulo/test/TableConfigurationUpdateIT.java
+++
b/test/src/main/java/org/apache/accumulo/test/TableConfigurationUpdateIT.java
@@ -78,14 +78,15 @@ public class TableConfigurationUpdateIT extends
AccumuloClusterHarness {
ExecutorService svc = Executors.newFixedThreadPool(numThreads);
CountDownLatch countDown = new CountDownLatch(numThreads);
ArrayList<Future<Exception>> futures = new ArrayList<>(numThreads);
-
- for (int i = 0; i < numThreads; i++) {
- futures.add(svc.submit(new TableConfRunner(randomMax, iterations,
tableConf, countDown)));
+ try {
+ for (int i = 0; i < numThreads; i++) {
+ futures.add(svc.submit(new TableConfRunner(randomMax, iterations,
tableConf, countDown)));
+ }
+ } finally {
+ svc.shutdown();
+ assertTrue(svc.awaitTermination(60, TimeUnit.MINUTES));
}
- svc.shutdown();
- assertTrue(svc.awaitTermination(60, TimeUnit.MINUTES));
-
for (Future<Exception> fut : futures) {
Exception e = fut.get();
if (e != null) {
diff --git a/test/src/main/java/org/apache/accumulo/test/TableOperationsIT.java
b/test/src/main/java/org/apache/accumulo/test/TableOperationsIT.java
index 21c835b0d3..610b2b7449 100644
--- a/test/src/main/java/org/apache/accumulo/test/TableOperationsIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/TableOperationsIT.java
@@ -212,44 +212,46 @@ public class TableOperationsIT extends
AccumuloClusterHarness {
final int initialTableSize =
accumuloClient.tableOperations().list().size();
final int numTasks = 10;
ExecutorService pool = Executors.newFixedThreadPool(numTasks);
+ try {
- for (String tablename : getUniqueNames(30)) {
- CountDownLatch startSignal = new CountDownLatch(numTasks);
- List<Future<Boolean>> futureList = new ArrayList<>(numTasks);
-
- for (int j = 0; j < numTasks; j++) {
- Future<Boolean> future = pool.submit(() -> {
- boolean result;
- try {
- startSignal.countDown();
- startSignal.await();
- accumuloClient.tableOperations().create(tablename);
- result = true;
- } catch (TableExistsException e) {
- result = false;
- }
- return result;
- });
- futureList.add(future);
- }
+ for (String tablename : getUniqueNames(30)) {
+ CountDownLatch startSignal = new CountDownLatch(numTasks);
+ List<Future<Boolean>> futureList = new ArrayList<>(numTasks);
+
+ for (int j = 0; j < numTasks; j++) {
+ Future<Boolean> future = pool.submit(() -> {
+ boolean result;
+ try {
+ startSignal.countDown();
+ startSignal.await();
+ accumuloClient.tableOperations().create(tablename);
+ result = true;
+ } catch (TableExistsException e) {
+ result = false;
+ }
+ return result;
+ });
+ futureList.add(future);
+ }
- int taskSucceeded = 0;
- int taskFailed = 0;
- for (Future<Boolean> result : futureList) {
- if (result.get() == true) {
- taskSucceeded++;
- } else {
- taskFailed++;
+ int taskSucceeded = 0;
+ int taskFailed = 0;
+ for (Future<Boolean> result : futureList) {
+ if (result.get() == true) {
+ taskSucceeded++;
+ } else {
+ taskFailed++;
+ }
}
+
+ assertEquals(1, taskSucceeded);
+ assertEquals(9, taskFailed);
}
- assertEquals(1, taskSucceeded);
- assertEquals(9, taskFailed);
+ assertEquals(30, accumuloClient.tableOperations().list().size() -
initialTableSize);
+ } finally {
+ pool.shutdown();
}
-
- assertEquals(30, accumuloClient.tableOperations().list().size() -
initialTableSize);
-
- pool.shutdown();
}
@Test
@@ -913,27 +915,29 @@ public class TableOperationsIT extends
AccumuloClusterHarness {
Set<TableId> hash = new HashSet<>();
ExecutorService pool = Executors.newFixedThreadPool(64);
+ try {
- for (int i = 0; i < 1000; i++) {
- int finalI = i;
+ for (int i = 0; i < 1000; i++) {
+ int finalI = i;
- Future<TableId> future = pool.submit(() -> {
- TableId tableId = null;
+ Future<TableId> future = pool.submit(() -> {
+ TableId tableId = null;
- tableId = Utils.getNextId("Testing" + finalI, getServerContext(),
TableId::of);
+ tableId = Utils.getNextId("Testing" + finalI, getServerContext(),
TableId::of);
- return tableId;
- });
+ return tableId;
+ });
- futureList.add(future);
- }
+ futureList.add(future);
+ }
- for (Future<TableId> tab : futureList) {
- hash.add(tab.get());
+ for (Future<TableId> tab : futureList) {
+ hash.add(tab.get());
+ }
+ } finally {
+ pool.shutdown();
}
- pool.shutdown();
-
assertEquals(1000, hash.size());
}
diff --git
a/test/src/main/java/org/apache/accumulo/test/UniqueNameAllocatorIT.java
b/test/src/main/java/org/apache/accumulo/test/UniqueNameAllocatorIT.java
index facbe5e1b5..5d8087b831 100644
--- a/test/src/main/java/org/apache/accumulo/test/UniqueNameAllocatorIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/UniqueNameAllocatorIT.java
@@ -69,53 +69,58 @@ class UniqueNameAllocatorIT extends SharedMiniClusterBase {
Set<String> namesSeen = ConcurrentHashMap.newKeySet();
var executorService = Executors.newCachedThreadPool();
- final int numLargeTasks = 64;
- final int numSmallTasks = 10;
- final int numTasks = numLargeTasks + numSmallTasks;
- List<Future<Integer>> futures = new ArrayList<>(numTasks);
- // start a portion of threads at the same time
- CountDownLatch startLatch = new CountDownLatch(32);
- assertTrue(numTasks >= startLatch.getCount(),
- "Not enough tasks to satisfy latch count - deadlock risk");
-
- // create threads that are allocating large random chunks
- for (int i = 0; i < numLargeTasks; i++) {
- var future = executorService.submit(() -> {
- startLatch.countDown();
- startLatch.await();
- int added = 0;
- while (namesSeen.size() < 1_000_000) {
- var allocator = allocators[random.nextInt(allocators.length)];
- int needed = Math.max(1, random.nextInt(999));
- allocate(allocator, needed, namesSeen);
- added += needed;
- }
- return added;
- });
- futures.add(future);
- }
+ try {
+ final int numLargeTasks = 64;
+ final int numSmallTasks = 10;
+ final int numTasks = numLargeTasks + numSmallTasks;
+ List<Future<Integer>> futures = new ArrayList<>(numTasks);
+ // start a portion of threads at the same time
+ CountDownLatch startLatch = new CountDownLatch(32);
+ assertTrue(numTasks >= startLatch.getCount(),
+ "Not enough tasks to satisfy latch count - deadlock risk");
- // create threads that are always allocating a small amount
- for (int i = 1; i <= numSmallTasks; i++) {
- int needed = i;
- var future = executorService.submit(() -> {
- startLatch.countDown();
- startLatch.await();
- int added = 0;
- while (namesSeen.size() < 1_000_000) {
- var allocator = allocators[random.nextInt(allocators.length)];
- allocate(allocator, needed, namesSeen);
- added += needed;
- }
- return added;
- });
- futures.add(future);
- }
- assertEquals(numTasks, futures.size());
+ // create threads that are allocating large random chunks
+ for (int i = 0; i < numLargeTasks; i++) {
+ var future = executorService.submit(() -> {
+ startLatch.countDown();
+ startLatch.await();
+ int added = 0;
+ while (namesSeen.size() < 1_000_000) {
+ var allocator = allocators[random.nextInt(allocators.length)];
+ int needed = Math.max(1, random.nextInt(999));
+ allocate(allocator, needed, namesSeen);
+ added += needed;
+ }
+ return added;
+ });
+ futures.add(future);
+ }
+
+ // create threads that are always allocating a small amount
+ for (int i = 1; i <= numSmallTasks; i++) {
+ int needed = i;
+ var future = executorService.submit(() -> {
+ startLatch.countDown();
+ startLatch.await();
+ int added = 0;
+ while (namesSeen.size() < 1_000_000) {
+ var allocator = allocators[random.nextInt(allocators.length)];
+ allocate(allocator, needed, namesSeen);
+ added += needed;
+ }
+ return added;
+ });
+ futures.add(future);
+ }
+ assertEquals(numTasks, futures.size());
+
+ for (var future : futures) {
+ // expect all threads to add some names
+ assertTrue(future.get() > 0);
+ }
- for (var future : futures) {
- // expect all threads to add some names
- assertTrue(future.get() > 0);
+ } finally {
+ executorService.shutdownNow();
}
assertThrows(IllegalArgumentException.class, () ->
allocators[0].getNextNames(-1));
diff --git a/test/src/main/java/org/apache/accumulo/test/ZombieScanIT.java
b/test/src/main/java/org/apache/accumulo/test/ZombieScanIT.java
index 80559f2415..60283f7423 100644
--- a/test/src/main/java/org/apache/accumulo/test/ZombieScanIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/ZombieScanIT.java
@@ -149,6 +149,7 @@ public class ZombieScanIT extends ConfigurableMacBase {
String table = getUniqueNames(1)[0];
+ var executor = Executors.newCachedThreadPool();
try (AccumuloClient c =
Accumulo.newClient().from(getClientProperties()).build()) {
var splits = new TreeSet<Text>();
@@ -170,8 +171,6 @@ public class ZombieScanIT extends ConfigurableMacBase {
// the in memory map it will wait for 15 seconds for the scan
c.tableOperations().flush(table, null, null, true);
- var executor = Executors.newCachedThreadPool();
-
// start two zombie scans that should never return using a normal scanner
List<Future<String>> futures = new ArrayList<>();
for (var row : List.of("2", "4")) {
@@ -252,6 +251,7 @@ public class ZombieScanIT extends ConfigurableMacBase {
// simple enough to include
assertValidScanIds(c);
+ } finally {
executor.shutdownNow();
}
@@ -273,12 +273,11 @@ public class ZombieScanIT extends ConfigurableMacBase {
final ServerType serverType = consistency == IMMEDIATE ? TABLET_SERVER :
SCAN_SERVER;
+ var executor = Executors.newCachedThreadPool();
try (AccumuloClient c =
Accumulo.newClient().from(getClientProperties()).build()) {
c.tableOperations().create(table);
- var executor = Executors.newCachedThreadPool();
-
// start four stuck scans that should never return data
List<Future<String>> futures = new ArrayList<>();
for (var row : List.of("2", "4")) {
@@ -337,6 +336,7 @@ public class ZombieScanIT extends ConfigurableMacBase {
assertEquals(6, countActiveScans(c, serverType, table));
+ } finally {
executor.shutdownNow();
}
}
diff --git
a/test/src/main/java/org/apache/accumulo/test/conf/PropStoreConfigIT_SimpleSuite.java
b/test/src/main/java/org/apache/accumulo/test/conf/PropStoreConfigIT_SimpleSuite.java
index 3039cbb3ed..c1bdbdbe9a 100644
---
a/test/src/main/java/org/apache/accumulo/test/conf/PropStoreConfigIT_SimpleSuite.java
+++
b/test/src/main/java/org/apache/accumulo/test/conf/PropStoreConfigIT_SimpleSuite.java
@@ -601,157 +601,158 @@ public class PropStoreConfigIT_SimpleSuite extends
SharedMiniClusterBase {
private static void runConcurrentPropsModificationTest(PropertyShim
propShim) throws Exception {
final int numTasks = 4;
ExecutorService executor = Executors.newFixedThreadPool(numTasks);
- CountDownLatch startLatch = new CountDownLatch(numTasks);
- assertTrue(numTasks >= startLatch.getCount(),
- "Not enough tasks/threads to satisfy latch count - deadlock risk");
- var tasks = new ArrayList<Callable<Void>>(numTasks);
+ try {
+ CountDownLatch startLatch = new CountDownLatch(numTasks);
+ assertTrue(numTasks >= startLatch.getCount(),
+ "Not enough tasks/threads to satisfy latch count - deadlock risk");
+ var tasks = new ArrayList<Callable<Void>>(numTasks);
+
+ final int iterations = 151;
+
+ tasks.add(() -> {
+ startLatch.countDown();
+ startLatch.await();
+ for (int i = 0; i < iterations; i++) {
+
+ Map<String,String> prevProps = null;
+ if (i % 10 == 0) {
+ prevProps = propShim.getProperties();
+ }
+
+ Map<String,String> acceptedProps =
propShim.modifyProperties(tableProps -> {
+ int A =
Integer.parseInt(tableProps.getOrDefault("general.custom.A", "0"));
+ int B =
Integer.parseInt(tableProps.getOrDefault("general.custom.B", "0"));
+ int C =
Integer.parseInt(tableProps.getOrDefault("general.custom.C", "0"));
+ int D =
Integer.parseInt(tableProps.getOrDefault("general.custom.D", "0"));
+
+ tableProps.put("general.custom.A", A + 2 + "");
+ tableProps.put("general.custom.B", B + 3 + "");
+ tableProps.put("general.custom.C", C + 5 + "");
+ tableProps.put("general.custom.D", D + 7 + "");
+ });
+
+ if (prevProps != null) {
+ var beforeA =
Integer.parseInt(prevProps.getOrDefault("general.custom.A", "0"));
+ var beforeB =
Integer.parseInt(prevProps.getOrDefault("general.custom.B", "0"));
+ var beforeC =
Integer.parseInt(prevProps.getOrDefault("general.custom.C", "0"));
+ var beforeD =
Integer.parseInt(prevProps.getOrDefault("general.custom.D", "0"));
+
+ var afterA =
Integer.parseInt(acceptedProps.get("general.custom.A"));
+ var afterB =
Integer.parseInt(acceptedProps.get("general.custom.B"));
+ var afterC =
Integer.parseInt(acceptedProps.get("general.custom.C"));
+ var afterD =
Integer.parseInt(acceptedProps.get("general.custom.D"));
+
+ // because there are other thread possibly making changes since
reading prevProps, can
+ // only do >= as opposed to == check. Should at a minimum see the
changes made by this
+ // thread.
+ assertTrue(afterA >= beforeA + 2);
+ assertTrue(afterB >= beforeB + 3);
+ assertTrue(afterC >= beforeC + 5);
+ assertTrue(afterD >= beforeD + 7);
+ }
+ }
+ return null;
+ });
- final int iterations = 151;
+ tasks.add(() -> {
+ startLatch.countDown();
+ startLatch.await();
+ for (int i = 0; i < iterations; i++) {
+ propShim.modifyProperties(tableProps -> {
+ int B =
Integer.parseInt(tableProps.getOrDefault("general.custom.B", "0"));
+ int C =
Integer.parseInt(tableProps.getOrDefault("general.custom.C", "0"));
+
+ tableProps.put("general.custom.B", B + 11 + "");
+ tableProps.put("general.custom.C", C + 13 + "");
+ });
+ }
+ return null;
+ });
- tasks.add(() -> {
- startLatch.countDown();
- startLatch.await();
- for (int i = 0; i < iterations; i++) {
+ tasks.add(() -> {
+ startLatch.countDown();
+ startLatch.await();
+ for (int i = 0; i < iterations; i++) {
+ propShim.modifyProperties(tableProps -> {
+ int B =
Integer.parseInt(tableProps.getOrDefault("general.custom.B", "0"));
- Map<String,String> prevProps = null;
- if (i % 10 == 0) {
- prevProps = propShim.getProperties();
+ tableProps.put("general.custom.B", B + 17 + "");
+ });
}
+ return null;
+ });
- Map<String,String> acceptedProps =
propShim.modifyProperties(tableProps -> {
- int A = Integer.parseInt(tableProps.getOrDefault("general.custom.A",
"0"));
- int B = Integer.parseInt(tableProps.getOrDefault("general.custom.B",
"0"));
- int C = Integer.parseInt(tableProps.getOrDefault("general.custom.C",
"0"));
- int D = Integer.parseInt(tableProps.getOrDefault("general.custom.D",
"0"));
-
- tableProps.put("general.custom.A", A + 2 + "");
- tableProps.put("general.custom.B", B + 3 + "");
- tableProps.put("general.custom.C", C + 5 + "");
- tableProps.put("general.custom.D", D + 7 + "");
- });
-
- if (prevProps != null) {
- var beforeA =
Integer.parseInt(prevProps.getOrDefault("general.custom.A", "0"));
- var beforeB =
Integer.parseInt(prevProps.getOrDefault("general.custom.B", "0"));
- var beforeC =
Integer.parseInt(prevProps.getOrDefault("general.custom.C", "0"));
- var beforeD =
Integer.parseInt(prevProps.getOrDefault("general.custom.D", "0"));
-
- var afterA = Integer.parseInt(acceptedProps.get("general.custom.A"));
- var afterB = Integer.parseInt(acceptedProps.get("general.custom.B"));
- var afterC = Integer.parseInt(acceptedProps.get("general.custom.C"));
- var afterD = Integer.parseInt(acceptedProps.get("general.custom.D"));
-
- // because there are other thread possibly making changes since
reading prevProps, can
- // only do >= as opposed to == check. Should at a minimum see the
changes made by this
- // thread.
- assertTrue(afterA >= beforeA + 2);
- assertTrue(afterB >= beforeB + 3);
- assertTrue(afterC >= beforeC + 5);
- assertTrue(afterD >= beforeD + 7);
+ tasks.add(() -> {
+ startLatch.countDown();
+ startLatch.await();
+ for (int i = 0; i < iterations; i++) {
+ propShim.modifyProperties(tableProps -> {
+ int E =
Integer.parseInt(tableProps.getOrDefault("general.custom.E", "0"));
+ tableProps.put("general.custom.E", E + 19 + "");
+ });
}
- }
- return null;
- });
-
- tasks.add(() -> {
- startLatch.countDown();
- startLatch.await();
- for (int i = 0; i < iterations; i++) {
- propShim.modifyProperties(tableProps -> {
- int B = Integer.parseInt(tableProps.getOrDefault("general.custom.B",
"0"));
- int C = Integer.parseInt(tableProps.getOrDefault("general.custom.C",
"0"));
-
- tableProps.put("general.custom.B", B + 11 + "");
- tableProps.put("general.custom.C", C + 13 + "");
- });
- }
- return null;
- });
+ return null;
+ });
- tasks.add(() -> {
- startLatch.countDown();
- startLatch.await();
- for (int i = 0; i < iterations; i++) {
- propShim.modifyProperties(tableProps -> {
- int B = Integer.parseInt(tableProps.getOrDefault("general.custom.B",
"0"));
+ assertEquals(numTasks, tasks.size());
- tableProps.put("general.custom.B", B + 17 + "");
- });
+ // run all of the above task concurrently
+ for (Future<Void> future : executor.invokeAll(tasks)) {
+ // see if there were any exceptions in the background thread and wait
for it to finish
+ future.get();
}
- return null;
- });
- tasks.add(() -> {
- startLatch.countDown();
- startLatch.await();
- for (int i = 0; i < iterations; i++) {
- propShim.modifyProperties(tableProps -> {
- int E = Integer.parseInt(tableProps.getOrDefault("general.custom.E",
"0"));
- tableProps.put("general.custom.E", E + 19 + "");
- });
- }
- return null;
- });
+ Map<String,String> expected = new HashMap<>();
+
+ // determine the expected sum for all the additions done by the separate
threads for each
+ // property
+ expected.put("general.custom.A", iterations * 2 + "");
+ expected.put("general.custom.B", iterations * (3 + 11 + 17) + "");
+ expected.put("general.custom.C", iterations * (5 + 13) + "");
+ expected.put("general.custom.D", iterations * 7 + "");
+ expected.put("general.custom.E", iterations * 19 + "");
+
+ final var IS_NOT_CUSTOM_TABLE_PROP =
+
Pattern.compile("general[.]custom[.][ABCDEF]").asMatchPredicate().negate();
+ Wait.waitFor(() -> {
+ var tableProps = new HashMap<>(propShim.getProperties());
+ tableProps.keySet().removeIf(IS_NOT_CUSTOM_TABLE_PROP);
+ boolean equal = expected.equals(tableProps);
+ if (!equal) {
+ log.info(
+ "Waiting for properties to converge. Actual:" + tableProps + "
Expected:" + expected);
+ }
+ return equal;
+ });
- assertEquals(numTasks, tasks.size());
+ // now that there are not other thread modifying properties, make a
modification to check that
+ // the returned map is exactly as expected.
+ Map<String,String> acceptedProps = propShim.modifyProperties(tableProps
-> {
+ int A = Integer.parseInt(tableProps.getOrDefault("general.custom.A",
"0"));
+ int B = Integer.parseInt(tableProps.getOrDefault("general.custom.B",
"0"));
+ int C = Integer.parseInt(tableProps.getOrDefault("general.custom.C",
"0"));
+ int D = Integer.parseInt(tableProps.getOrDefault("general.custom.D",
"0"));
+
+ tableProps.put("general.custom.A", A + 2 + "");
+ tableProps.put("general.custom.B", B + 3 + "");
+ tableProps.put("general.custom.C", C + 5 + "");
+ tableProps.put("general.custom.D", D + 7 + "");
+ });
- // run all of the above task concurrently
- for (Future<Void> future : executor.invokeAll(tasks)) {
- // see if there were any exceptions in the background thread and wait
for it to finish
- future.get();
+ var afterA = Integer.parseInt(acceptedProps.get("general.custom.A"));
+ var afterB = Integer.parseInt(acceptedProps.get("general.custom.B"));
+ var afterC = Integer.parseInt(acceptedProps.get("general.custom.C"));
+ var afterD = Integer.parseInt(acceptedProps.get("general.custom.D"));
+ var afterE = Integer.parseInt(acceptedProps.get("general.custom.E"));
+
+ assertEquals(iterations * 2 + 2, afterA);
+ assertEquals(iterations * (3 + 11 + 17) + 3, afterB);
+ assertEquals(iterations * (5 + 13) + 5, afterC);
+ assertEquals(iterations * 7 + 7, afterD);
+ assertEquals(iterations * 19, afterE);
+ } finally {
+ executor.shutdown();
}
-
- Map<String,String> expected = new HashMap<>();
-
- // determine the expected sum for all the additions done by the separate
threads for each
- // property
- expected.put("general.custom.A", iterations * 2 + "");
- expected.put("general.custom.B", iterations * (3 + 11 + 17) + "");
- expected.put("general.custom.C", iterations * (5 + 13) + "");
- expected.put("general.custom.D", iterations * 7 + "");
- expected.put("general.custom.E", iterations * 19 + "");
-
- final var IS_NOT_CUSTOM_TABLE_PROP =
-
Pattern.compile("general[.]custom[.][ABCDEF]").asMatchPredicate().negate();
- Wait.waitFor(() -> {
- var tableProps = new HashMap<>(propShim.getProperties());
- tableProps.keySet().removeIf(IS_NOT_CUSTOM_TABLE_PROP);
- boolean equal = expected.equals(tableProps);
- if (!equal) {
- log.info(
- "Waiting for properties to converge. Actual:" + tableProps + "
Expected:" + expected);
- }
- return equal;
- });
-
- // now that there are not other thread modifying properties, make a
modification to check that
- // the returned map
- // is exactly as expected.
- Map<String,String> acceptedProps = propShim.modifyProperties(tableProps ->
{
- int A = Integer.parseInt(tableProps.getOrDefault("general.custom.A",
"0"));
- int B = Integer.parseInt(tableProps.getOrDefault("general.custom.B",
"0"));
- int C = Integer.parseInt(tableProps.getOrDefault("general.custom.C",
"0"));
- int D = Integer.parseInt(tableProps.getOrDefault("general.custom.D",
"0"));
-
- tableProps.put("general.custom.A", A + 2 + "");
- tableProps.put("general.custom.B", B + 3 + "");
- tableProps.put("general.custom.C", C + 5 + "");
- tableProps.put("general.custom.D", D + 7 + "");
- });
-
- var afterA = Integer.parseInt(acceptedProps.get("general.custom.A"));
- var afterB = Integer.parseInt(acceptedProps.get("general.custom.B"));
- var afterC = Integer.parseInt(acceptedProps.get("general.custom.C"));
- var afterD = Integer.parseInt(acceptedProps.get("general.custom.D"));
- var afterE = Integer.parseInt(acceptedProps.get("general.custom.E"));
-
- assertEquals(iterations * 2 + 2, afterA);
- assertEquals(iterations * (3 + 11 + 17) + 3, afterB);
- assertEquals(iterations * (5 + 13) + 5, afterC);
- assertEquals(iterations * 7 + 7, afterD);
- assertEquals(iterations * 19, afterE);
-
- executor.shutdown();
}
@Test
diff --git
a/test/src/main/java/org/apache/accumulo/test/fate/meta/ZooMutatorIT.java
b/test/src/main/java/org/apache/accumulo/test/fate/meta/ZooMutatorIT.java
index f0ad481e6b..d689db45fa 100644
--- a/test/src/main/java/org/apache/accumulo/test/fate/meta/ZooMutatorIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/fate/meta/ZooMutatorIT.java
@@ -86,12 +86,11 @@ public class ZooMutatorIT extends WithTestNames {
if (!Files.isDirectory(newFolder)) {
Files.createDirectories(newFolder);
}
+ final int numTasks = 16;
+ var executor = Executors.newFixedThreadPool(numTasks);
try (var testZk = new ZooKeeperTestingServer(newFolder.toFile()); var zk =
testZk.newClient()) {
var zrw = zk.asReaderWriter();
- final int numTasks = 16;
- var executor = Executors.newFixedThreadPool(numTasks);
-
String initialData = hash("Accumulo Zookeeper Mutator test data") + " 0";
List<Future<List<Integer>>> futures = new ArrayList<>(numTasks);
@@ -131,7 +130,6 @@ public class ZooMutatorIT extends WithTestNames {
countCounts.put(count, countCounts.getOrDefault(count, 0) + 1);
}
}
- executor.shutdown();
byte[] actual = zrw.getData("/test-zm");
int settledCount = getCount(actual);
@@ -149,6 +147,8 @@ public class ZooMutatorIT extends WithTestNames {
assertEquals(settledCount + 1, countCounts.size());
assertEquals(expected, new String(actual, UTF_8));
+ } finally {
+ executor.shutdown();
}
}
diff --git
a/test/src/main/java/org/apache/accumulo/test/functional/AddSplitIT_SimpleSuite.java
b/test/src/main/java/org/apache/accumulo/test/functional/AddSplitIT_SimpleSuite.java
index 52cd4550f8..b85679b115 100644
---
a/test/src/main/java/org/apache/accumulo/test/functional/AddSplitIT_SimpleSuite.java
+++
b/test/src/main/java/org/apache/accumulo/test/functional/AddSplitIT_SimpleSuite.java
@@ -291,7 +291,6 @@ public class AddSplitIT_SimpleSuite extends
SharedMiniClusterBase {
@Test
public void concurrentAddSplitTest() throws Exception {
var threads = 10;
- var service = Executors.newFixedThreadPool(threads);
var latch = new CountDownLatch(threads);
String tableName = getUniqueNames(1)[0];
@@ -311,37 +310,40 @@ public class AddSplitIT_SimpleSuite extends
SharedMiniClusterBase {
var commonSplits = commonBuilder.build();
allSplits.putAll(commonSplits);
- // Spin up 10 threads and concurrently submit all 50 existing splits
- // as well as 50 unique splits, this should create a collision with fate
- // and cause retries
- for (int i = 1; i <= threads; i++) {
- var start = i * 100;
- service.execute(() -> {
- // add the 50 common splits
- SortedMap<Text,TabletMergeability> splits = new
TreeMap<>(commonSplits);
- // create 50 unique splits
- for (int j = start; j < start + 50; j++) {
- splits.put(new Text(String.format("%09d", j)),
- TabletMergeability.after(Duration.ofHours(1 + j)));
- }
- // make sure all splits are captured
- allSplits.putAll(splits);
- // Wait for all 10 threads to be ready before calling putSplits()
- // to increase the chance of collisions
- latch.countDown();
- try {
- latch.await();
- c.tableOperations().putSplits(tableName, splits);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- });
+ var service = Executors.newFixedThreadPool(threads);
+ try {
+ // Spin up 10 threads and concurrently submit all 50 existing splits
+ // as well as 50 unique splits, this should create a collision with
fate
+ // and cause retries
+ for (int i = 1; i <= threads; i++) {
+ var start = i * 100;
+ service.execute(() -> {
+ // add the 50 common splits
+ SortedMap<Text,TabletMergeability> splits = new
TreeMap<>(commonSplits);
+ // create 50 unique splits
+ for (int j = start; j < start + 50; j++) {
+ splits.put(new Text(String.format("%09d", j)),
+ TabletMergeability.after(Duration.ofHours(1 + j)));
+ }
+ // make sure all splits are captured
+ allSplits.putAll(splits);
+ // Wait for all 10 threads to be ready before calling putSplits()
+ // to increase the chance of collisions
+ latch.countDown();
+ try {
+ latch.await();
+ c.tableOperations().putSplits(tableName, splits);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+ } finally {
+ // Wait for all 10 threads to finish
+ service.shutdown();
+ assertTrue(service.awaitTermination(2, TimeUnit.MINUTES));
}
- // Wait for all 10 threads to finish
- service.shutdown();
- assertTrue(service.awaitTermination(2, TimeUnit.MINUTES));
-
// Verify we have 550 splits and then all splits are correctly set
assertEquals(50 + (threads * 50), allSplits.size());
TableId id = TableId.of(c.tableOperations().tableIdMap().get(tableName));
diff --git
a/test/src/main/java/org/apache/accumulo/test/functional/BulkNewIT.java
b/test/src/main/java/org/apache/accumulo/test/functional/BulkNewIT.java
index 36ed26a3fb..474b8edc11 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/BulkNewIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/BulkNewIT.java
@@ -470,41 +470,45 @@ public class BulkNewIT extends SharedMiniClusterBase {
// Start a second bulk import in background thread because it is
expected this bulk import
// will hang because tablets are over the pause file limit.
ExecutorService executor = Executors.newFixedThreadPool(1);
- var future = executor.submit(() -> {
-
client.tableOperations().importDirectory(dir2).to(tableName).tableTime(true).load();
- return null;
- });
-
- // sleep a bit to give the bulk import a chance to run
- UtilWaitThread.sleep(3000);
- // bulk import should not have gone through it should be pausing because
the tablet have too
- // many files
- assertFalse(future.isDone());
- verifyData(client, tableName, 0, 179, false);
+ try {
+ var future = executor.submit(() -> {
+
client.tableOperations().importDirectory(dir2).to(tableName).tableTime(true).load();
+ return null;
+ });
- // Before the bulk import runs no tablets should have loaded flags set
- assertEquals(Map.of("0060", 0, "0120", 0, "null", 0),
countLoaded(client, tableName));
- // compacting the first tablet should allow the import on that tablet to
proceed
- client.tableOperations().compact(tableName,
- new CompactionConfig().setWait(true).setEndRow(new Text("0060")));
- // Wait for the first tablets data to be updated by bulk import.
- Wait.waitFor(
- () -> Map.of("0060", 7, "0120", 0, "null",
0).equals(countLoaded(client, tableName)));
-
- // The bulk imports on the other tablets should not have gone through,
verify their data was
- // not updated. Spot check a few rows in the other two tablets. The
first tablet may or may
- // not be updated on the tablet server at this point, so can not look at
its data.
- assertEquals(61L, readRowValue(client, tableName, 61));
- assertEquals(100L, readRowValue(client, tableName, 100));
- assertEquals(140L, readRowValue(client, tableName, 140));
-
- // compact the entire table, should allow all bulk imports to go through
- client.tableOperations().compact(tableName, new
CompactionConfig().setWait(true));
- // wait for bulk import to complete
- future.get();
- // verify the values were updated by the bulk import that was paused
- verifyData(client, tableName, 0, 179, 1000, false);
- assertEquals(Map.of("0060", 0, "0120", 0, "null", 0),
countLoaded(client, tableName));
+ // sleep a bit to give the bulk import a chance to run
+ UtilWaitThread.sleep(3000);
+ // bulk import should not have gone through it should be pausing
because the tablet have too
+ // many files
+ assertFalse(future.isDone());
+ verifyData(client, tableName, 0, 179, false);
+
+ // Before the bulk import runs no tablets should have loaded flags set
+ assertEquals(Map.of("0060", 0, "0120", 0, "null", 0),
countLoaded(client, tableName));
+ // compacting the first tablet should allow the import on that tablet
to proceed
+ client.tableOperations().compact(tableName,
+ new CompactionConfig().setWait(true).setEndRow(new Text("0060")));
+ // Wait for the first tablets data to be updated by bulk import.
+ Wait.waitFor(
+ () -> Map.of("0060", 7, "0120", 0, "null",
0).equals(countLoaded(client, tableName)));
+
+ // The bulk imports on the other tablets should not have gone through,
verify their data was
+ // not updated. Spot check a few rows in the other two tablets. The
first tablet may or may
+ // not be updated on the tablet server at this point, so can not look
at its data.
+ assertEquals(61L, readRowValue(client, tableName, 61));
+ assertEquals(100L, readRowValue(client, tableName, 100));
+ assertEquals(140L, readRowValue(client, tableName, 140));
+
+ // compact the entire table, should allow all bulk imports to go
through
+ client.tableOperations().compact(tableName, new
CompactionConfig().setWait(true));
+ // wait for bulk import to complete
+ future.get();
+ // verify the values were updated by the bulk import that was paused
+ verifyData(client, tableName, 0, 179, 1000, false);
+ assertEquals(Map.of("0060", 0, "0120", 0, "null", 0),
countLoaded(client, tableName));
+ } finally {
+ executor.shutdown();
+ }
}
}
@@ -1200,7 +1204,6 @@ public class BulkNewIT extends SharedMiniClusterBase {
c.tableOperations().addSplits(tableName, splits);
final int numTasks = 16;
- var executor = Executors.newFixedThreadPool(numTasks);
var futures = new ArrayList<Future<?>>();
// wait for a portion of the tasks to be ready
CountDownLatch startLatch = new CountDownLatch(numTasks);
@@ -1212,27 +1215,30 @@ public class BulkNewIT extends SharedMiniClusterBase {
var imports = IntStream.range(2,
8999).boxed().collect(Collectors.toList());
// The order in which imports are added to the load plan should not
matter so test that.
Collections.shuffle(imports);
- for (var data : imports) {
- String filename = "f" + data + ".";
- loadPlanBuilder.loadFileTo(filename + RFile.EXTENSION,
RangeType.TABLE, row(data - 1),
- row(data));
- var future = executor.submit(() -> {
- startLatch.countDown();
- startLatch.await();
- writeData(fs, dir + "/" + filename, aconf, data, data);
- return null;
- });
- futures.add(future);
- rowsExpected.add(row(data));
- }
- assertEquals(imports.size(), futures.size());
+ var executor = Executors.newFixedThreadPool(numTasks);
+ try {
+ for (var data : imports) {
+ String filename = "f" + data + ".";
+ loadPlanBuilder.loadFileTo(filename + RFile.EXTENSION,
RangeType.TABLE, row(data - 1),
+ row(data));
+ var future = executor.submit(() -> {
+ startLatch.countDown();
+ startLatch.await();
+ writeData(fs, dir + "/" + filename, aconf, data, data);
+ return null;
+ });
+ futures.add(future);
+ rowsExpected.add(row(data));
+ }
+ assertEquals(imports.size(), futures.size());
- for (var future : futures) {
- future.get();
+ for (var future : futures) {
+ future.get();
+ }
+ } finally {
+ executor.shutdown();
}
- executor.shutdown();
-
var loadPlan = loadPlanBuilder.build();
c.tableOperations().importDirectory(dir).to(tableName).plan(loadPlan).load();
diff --git
a/test/src/main/java/org/apache/accumulo/test/functional/CompactionIT.java
b/test/src/main/java/org/apache/accumulo/test/functional/CompactionIT.java
index 2e6fbfa7d9..000ffd26e0 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/CompactionIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/CompactionIT.java
@@ -286,26 +286,29 @@ public class CompactionIT extends CompactionITBase {
final int THREADS = 5;
for (int count = 0; count < THREADS; count++) {
ExecutorService executor = Executors.newFixedThreadPool(THREADS);
- final int span = 500000 / 59;
- for (int i = 0; i < 500000; i += 500000 / 59) {
- final int finalI = i;
- Runnable r = () -> {
- try {
- VerifyParams params = new VerifyParams(getClientProps(),
tableName, span);
- params.startRow = finalI;
- params.random = 56;
- params.dataSize = 50;
- params.cols = 1;
- VerifyIngest.verifyIngest(c, params);
- } catch (Exception ex) {
- log.warn("Got exception verifying data", ex);
- fail.set(true);
- }
- };
- executor.execute(r);
+ try {
+ final int span = 500000 / 59;
+ for (int i = 0; i < 500000; i += 500000 / 59) {
+ final int finalI = i;
+ Runnable r = () -> {
+ try {
+ VerifyParams params = new VerifyParams(getClientProps(),
tableName, span);
+ params.startRow = finalI;
+ params.random = 56;
+ params.dataSize = 50;
+ params.cols = 1;
+ VerifyIngest.verifyIngest(c, params);
+ } catch (Exception ex) {
+ log.warn("Got exception verifying data", ex);
+ fail.set(true);
+ }
+ };
+ executor.execute(r);
+ }
+ } finally {
+ executor.shutdown();
+ executor.awaitTermination(defaultTimeout().toSeconds(), SECONDS);
}
- executor.shutdown();
- executor.awaitTermination(defaultTimeout().toSeconds(), SECONDS);
assertFalse(fail.get(),
"Failed to successfully run all threads, Check the test output for
error");
}
@@ -1070,56 +1073,59 @@ public class CompactionIT extends CompactionITBase {
client.tableOperations().setProperty(table,
Property.TABLE_MAJC_RATIO.getKey(), "1");
// start a bunch of compactions in the background
- var executor = Executors.newCachedThreadPool();
final int numTasks = 20;
List<Future<?>> futures = new ArrayList<>(numTasks);
CountDownLatch startLatch = new CountDownLatch(numTasks);
assertTrue(numTasks >= startLatch.getCount(),
"Not enough tasks/threads to satisfy latch count - deadlock risk");
- // start user compactions on a subset of the tables tablets, system
compactions should attempt
- // to run on all tablets. With concurrency should get a mix.
- for (int i = 1; i < numTasks + 1; i++) {
- var startRow = new Text(String.format("r:%04d", i - 1));
- var endRow = new Text(String.format("r:%04d", i));
- final CompactionConfig config = new CompactionConfig();
- config.setWait(true);
- config.setStartRow(startRow);
- config.setEndRow(endRow);
- futures.add(executor.submit(() -> {
- startLatch.countDown();
- startLatch.await();
- client.tableOperations().compact(table, config);
- return null;
- }));
- }
- assertEquals(numTasks, futures.size());
+ var executor = Executors.newCachedThreadPool();
+ try {
+ // start user compactions on a subset of the tables tablets, system
compactions should
+ // attempt to run on all tablets. With concurrency should get a mix.
+ for (int i = 1; i < numTasks + 1; i++) {
+ var startRow = new Text(String.format("r:%04d", i - 1));
+ var endRow = new Text(String.format("r:%04d", i));
+ final CompactionConfig config = new CompactionConfig();
+ config.setWait(true);
+ config.setStartRow(startRow);
+ config.setEndRow(endRow);
+ futures.add(executor.submit(() -> {
+ startLatch.countDown();
+ startLatch.await();
+ client.tableOperations().compact(table, config);
+ return null;
+ }));
+ }
+ assertEquals(numTasks, futures.size());
- log.debug("Waiting for offline");
- // take tablet offline while there are concurrent compactions
- client.tableOperations().offline(table, true);
+ log.debug("Waiting for offline");
+ // take tablet offline while there are concurrent compactions
+ client.tableOperations().offline(table, true);
- // grab a snapshot of all the tablets files after waiting for offline,
do not expect any
- // tablets files to change at this point
- var files1 = getFiles(ctx, tableId);
+ // grab a snapshot of all the tablets files after waiting for offline,
do not expect any
+ // tablets files to change at this point
+ var files1 = getFiles(ctx, tableId);
- // wait for the background compactions
- log.debug("Waiting for futures");
- for (var future : futures) {
- try {
- future.get();
- } catch (ExecutionException ee) {
- // its ok if some of the compactions fail because the table was
concurrently taken offline
- assertTrue(ee.getMessage().contains("is offline"));
+ // wait for the background compactions
+ log.debug("Waiting for futures");
+ for (var future : futures) {
+ try {
+ future.get();
+ } catch (ExecutionException ee) {
+ // its ok if some of the compactions fail because the table was
concurrently taken
+ // offline
+ assertTrue(ee.getMessage().contains("is offline"));
+ }
}
- }
- // grab a second snapshot of the tablets files after all the background
operations completed
- var files2 = getFiles(ctx, tableId);
+ // grab a second snapshot of the tablets files after all the
background operations completed
+ var files2 = getFiles(ctx, tableId);
- // do not expect the files to have changed after the offline operation
returned.
- assertEquals(files1, files2);
-
- executor.shutdown();
+ // do not expect the files to have changed after the offline operation
returned.
+ assertEquals(files1, files2);
+ } finally {
+ executor.shutdown();
+ }
}
}
diff --git
a/test/src/main/java/org/apache/accumulo/test/functional/ConcurrentDeleteTableIT.java
b/test/src/main/java/org/apache/accumulo/test/functional/ConcurrentDeleteTableIT.java
index 275223b192..64de2ab6cd 100644
---
a/test/src/main/java/org/apache/accumulo/test/functional/ConcurrentDeleteTableIT.java
+++
b/test/src/main/java/org/apache/accumulo/test/functional/ConcurrentDeleteTableIT.java
@@ -65,13 +65,13 @@ public class ConcurrentDeleteTableIT extends
AccumuloClusterHarness {
@Test
public void testConcurrentDeleteTablesOps() throws Exception {
- try (AccumuloClient c =
Accumulo.newClient().from(getClientProps()).build()) {
+ int numDeleteOps = 20;
- String[] tables = getUniqueNames(NUM_TABLES);
+ ExecutorService es = Executors.newFixedThreadPool(numDeleteOps);
- int numDeleteOps = 20;
+ try (AccumuloClient c =
Accumulo.newClient().from(getClientProps()).build()) {
- ExecutorService es = Executors.newFixedThreadPool(numDeleteOps);
+ String[] tables = getUniqueNames(NUM_TABLES);
int count = 0;
for (final String table : tables) {
@@ -119,19 +119,21 @@ public class ConcurrentDeleteTableIT extends
AccumuloClusterHarness {
FunctionalTestUtils.assertNoDanglingFateLocks(getCluster());
}
- es.shutdown();
+ } finally {
+ es.shutdownNow();
}
}
@Test
public void testConcurrentFateOpsWithDelete() throws Exception {
- try (AccumuloClient c =
Accumulo.newClient().from(getClientProps()).build()) {
- String[] tables = getUniqueNames(NUM_TABLES);
+ int numOperations = 8;
+
+ ExecutorService es = Executors.newFixedThreadPool(numOperations);
- int numOperations = 8;
+ try (AccumuloClient c =
Accumulo.newClient().from(getClientProps()).build()) {
- ExecutorService es = Executors.newFixedThreadPool(numOperations);
+ String[] tables = getUniqueNames(NUM_TABLES);
int count = 0;
for (final String table : tables) {
@@ -227,7 +229,8 @@ public class ConcurrentDeleteTableIT extends
AccumuloClusterHarness {
FunctionalTestUtils.assertNoDanglingFateLocks(getCluster());
}
- es.shutdown();
+ } finally {
+ es.shutdownNow();
}
}
diff --git
a/test/src/main/java/org/apache/accumulo/test/functional/ConcurrentTableNameOperationsIT.java
b/test/src/main/java/org/apache/accumulo/test/functional/ConcurrentTableNameOperationsIT.java
index 571fbbd334..ea5642ac5a 100644
---
a/test/src/main/java/org/apache/accumulo/test/functional/ConcurrentTableNameOperationsIT.java
+++
b/test/src/main/java/org/apache/accumulo/test/functional/ConcurrentTableNameOperationsIT.java
@@ -88,31 +88,33 @@ public class ConcurrentTableNameOperationsIT extends
SharedMiniClusterBase {
final int numTasks = 16;
final int numIterations = 8;
ExecutorService pool = Executors.newFixedThreadPool(numTasks);
+ try {
+
+ for (String targetTableName : getUniqueNames(numIterations)) {
+ List<String> sourceTableNames = new ArrayList<>();
+ for (int i = 0; i < numTasks; i++) {
+ String sourceTable = targetTableName + "_source_" + i;
+ client.tableOperations().create(sourceTable);
+ sourceTableNames.add(sourceTable);
+ }
- for (String targetTableName : getUniqueNames(numIterations)) {
- List<String> sourceTableNames = new ArrayList<>();
- for (int i = 0; i < numTasks; i++) {
- String sourceTable = targetTableName + "_source_" + i;
- client.tableOperations().create(sourceTable);
- sourceTableNames.add(sourceTable);
- }
-
- int tableCountBefore = client.tableOperations().list().size();
+ int tableCountBefore = client.tableOperations().list().size();
- int successCount = runConcurrentTableOperation(pool, numTasks, (index)
-> {
- client.tableOperations().clone(sourceTableNames.get(index),
targetTableName, true, Map.of(),
- Set.of());
- return true;
- });
+ int successCount = runConcurrentTableOperation(pool, numTasks, (index)
-> {
+ client.tableOperations().clone(sourceTableNames.get(index),
targetTableName, true,
+ Map.of(), Set.of());
+ return true;
+ });
- assertEquals(1, successCount, "Expected only one clone operation to
succeed");
- assertTrue(client.tableOperations().exists(targetTableName),
- "Expected target table " + targetTableName + " to exist");
- assertEquals(tableCountBefore + 1,
client.tableOperations().list().size(),
- "Expected only one new table after clone");
+ assertEquals(1, successCount, "Expected only one clone operation to
succeed");
+ assertTrue(client.tableOperations().exists(targetTableName),
+ "Expected target table " + targetTableName + " to exist");
+ assertEquals(tableCountBefore + 1,
client.tableOperations().list().size(),
+ "Expected only one new table after clone");
+ }
+ } finally {
+ pool.shutdown();
}
-
- pool.shutdown();
}
/**
@@ -123,29 +125,31 @@ public class ConcurrentTableNameOperationsIT extends
SharedMiniClusterBase {
final int numTasks = 16;
final int numIterations = 10;
ExecutorService pool = Executors.newFixedThreadPool(numTasks);
+ try {
+
+ for (String targetTableName : getUniqueNames(numIterations)) {
+ List<String> sourceTableNames = new ArrayList<>();
+ for (int i = 0; i < numTasks; i++) {
+ String sourceTable = targetTableName + "_rename_source_" + i;
+ client.tableOperations().create(sourceTable);
+ sourceTableNames.add(sourceTable);
+ }
- for (String targetTableName : getUniqueNames(numIterations)) {
- List<String> sourceTableNames = new ArrayList<>();
- for (int i = 0; i < numTasks; i++) {
- String sourceTable = targetTableName + "_rename_source_" + i;
- client.tableOperations().create(sourceTable);
- sourceTableNames.add(sourceTable);
- }
-
- int tableCountBefore = client.tableOperations().list().size();
+ int tableCountBefore = client.tableOperations().list().size();
- int successCount = runConcurrentTableOperation(pool, numTasks, (index)
-> {
- client.tableOperations().rename(sourceTableNames.get(index),
targetTableName);
- return true;
- });
+ int successCount = runConcurrentTableOperation(pool, numTasks, (index)
-> {
+ client.tableOperations().rename(sourceTableNames.get(index),
targetTableName);
+ return true;
+ });
- assertEquals(1, successCount, "Expected only one rename operation to
succeed");
- assertTrue(client.tableOperations().exists(targetTableName),
- "Expected target table " + targetTableName + " to exist");
- assertEquals(tableCountBefore, client.tableOperations().list().size());
+ assertEquals(1, successCount, "Expected only one rename operation to
succeed");
+ assertTrue(client.tableOperations().exists(targetTableName),
+ "Expected target table " + targetTableName + " to exist");
+ assertEquals(tableCountBefore, client.tableOperations().list().size());
+ }
+ } finally {
+ pool.shutdown();
}
-
- pool.shutdown();
}
/**
@@ -157,36 +161,38 @@ public class ConcurrentTableNameOperationsIT extends
SharedMiniClusterBase {
final int numTasks = 16;
final int numIterations = 4;
ExecutorService pool = Executors.newFixedThreadPool(numTasks);
- String[] targetTableNames = getUniqueNames(numIterations);
- var ntc = new NewTableConfiguration().createOffline();
-
- for (String importTableName : targetTableNames) {
- // Create separate source tables and export directories for each thread
- List<String> exportDirs = new ArrayList<>(numTasks);
- for (int i = 0; i < numTasks; i++) {
- String sourceTableName = importTableName + "_export_source_" + i;
- client.tableOperations().create(sourceTableName, ntc);
- String exportDir = getCluster().getTemporaryPath() + "/export_" +
sourceTableName;
- client.tableOperations().exportTable(sourceTableName, exportDir);
- exportDirs.add(exportDir);
- }
+ try {
+ String[] targetTableNames = getUniqueNames(numIterations);
+ var ntc = new NewTableConfiguration().createOffline();
+
+ for (String importTableName : targetTableNames) {
+ // Create separate source tables and export directories for each thread
+ List<String> exportDirs = new ArrayList<>(numTasks);
+ for (int i = 0; i < numTasks; i++) {
+ String sourceTableName = importTableName + "_export_source_" + i;
+ client.tableOperations().create(sourceTableName, ntc);
+ String exportDir = getCluster().getTemporaryPath() + "/export_" +
sourceTableName;
+ client.tableOperations().exportTable(sourceTableName, exportDir);
+ exportDirs.add(exportDir);
+ }
- int tableCountBefore = client.tableOperations().list().size();
+ int tableCountBefore = client.tableOperations().list().size();
- // All threads attempt to import to the same target table name
- int successCount = runConcurrentTableOperation(pool, numTasks, (index)
-> {
- client.tableOperations().importTable(importTableName,
exportDirs.get(index));
- return true;
- });
+ // All threads attempt to import to the same target table name
+ int successCount = runConcurrentTableOperation(pool, numTasks, (index)
-> {
+ client.tableOperations().importTable(importTableName,
exportDirs.get(index));
+ return true;
+ });
- assertEquals(1, successCount, "Expected only one import operation to
succeed");
- assertTrue(client.tableOperations().exists(importTableName),
- "Expected import table " + importTableName + " to exist");
- assertEquals(tableCountBefore + 1,
client.tableOperations().list().size(),
- "Expected +1 table count for import operation");
+ assertEquals(1, successCount, "Expected only one import operation to
succeed");
+ assertTrue(client.tableOperations().exists(importTableName),
+ "Expected import table " + importTableName + " to exist");
+ assertEquals(tableCountBefore + 1,
client.tableOperations().list().size(),
+ "Expected +1 table count for import operation");
+ }
+ } finally {
+ pool.shutdown();
}
-
- pool.shutdown();
}
/**
@@ -199,94 +205,96 @@ public class ConcurrentTableNameOperationsIT extends
SharedMiniClusterBase {
final int numTasks = operationsPerType * 3;
final int numIterations = 4;
ExecutorService pool = Executors.newFixedThreadPool(numTasks);
- String[] expectedTableNames = getUniqueNames(numIterations);
-
- for (String targetTableName : expectedTableNames) {
- List<String> cloneSourceTables = new ArrayList<>();
- List<String> renameSourceTables = new ArrayList<>();
- for (int i = 0; i < operationsPerType; i++) {
- String cloneSource = targetTableName + "_clone_src_" + i;
- client.tableOperations().create(cloneSource);
- cloneSourceTables.add(cloneSource);
-
- String renameSource = targetTableName + "_rename_src_" + i;
- client.tableOperations().create(renameSource);
- renameSourceTables.add(renameSource);
- }
+ try {
+ String[] expectedTableNames = getUniqueNames(numIterations);
+
+ for (String targetTableName : expectedTableNames) {
+ List<String> cloneSourceTables = new ArrayList<>();
+ List<String> renameSourceTables = new ArrayList<>();
+ for (int i = 0; i < operationsPerType; i++) {
+ String cloneSource = targetTableName + "_clone_src_" + i;
+ client.tableOperations().create(cloneSource);
+ cloneSourceTables.add(cloneSource);
+
+ String renameSource = targetTableName + "_rename_src_" + i;
+ client.tableOperations().create(renameSource);
+ renameSourceTables.add(renameSource);
+ }
- int tableCountBefore = client.tableOperations().list().size();
-
- List<Future<Boolean>> futures = new ArrayList<>();
- CountDownLatch startSignal = new CountDownLatch(numTasks);
- AtomicReference<String> successfulOperation = new AtomicReference<>();
-
- for (int i = 0; i < operationsPerType; i++) {
- futures.add(pool.submit(() -> {
- try {
- startSignal.countDown();
- startSignal.await();
- client.tableOperations().create(targetTableName);
- successfulOperation.set("create");
- return true;
- } catch (TableExistsException e) {
- return false;
- }
- }));
-
- final int index = i;
-
- futures.add(pool.submit(() -> {
- try {
- startSignal.countDown();
- startSignal.await();
- client.tableOperations().rename(renameSourceTables.get(index),
targetTableName);
- successfulOperation.set("rename");
- return true;
- } catch (TableExistsException e) {
- return false;
- }
- }));
-
- futures.add(pool.submit(() -> {
- try {
- startSignal.countDown();
- startSignal.await();
- client.tableOperations().clone(cloneSourceTables.get(index),
targetTableName, true,
- Map.of(), Set.of());
- successfulOperation.set("clone");
- return true;
- } catch (TableExistsException e) {
- return false;
- }
- }));
- }
+ int tableCountBefore = client.tableOperations().list().size();
+
+ List<Future<Boolean>> futures = new ArrayList<>();
+ CountDownLatch startSignal = new CountDownLatch(numTasks);
+ AtomicReference<String> successfulOperation = new AtomicReference<>();
+
+ for (int i = 0; i < operationsPerType; i++) {
+ futures.add(pool.submit(() -> {
+ try {
+ startSignal.countDown();
+ startSignal.await();
+ client.tableOperations().create(targetTableName);
+ successfulOperation.set("create");
+ return true;
+ } catch (TableExistsException e) {
+ return false;
+ }
+ }));
+
+ final int index = i;
+
+ futures.add(pool.submit(() -> {
+ try {
+ startSignal.countDown();
+ startSignal.await();
+ client.tableOperations().rename(renameSourceTables.get(index),
targetTableName);
+ successfulOperation.set("rename");
+ return true;
+ } catch (TableExistsException e) {
+ return false;
+ }
+ }));
+
+ futures.add(pool.submit(() -> {
+ try {
+ startSignal.countDown();
+ startSignal.await();
+ client.tableOperations().clone(cloneSourceTables.get(index),
targetTableName, true,
+ Map.of(), Set.of());
+ successfulOperation.set("clone");
+ return true;
+ } catch (TableExistsException e) {
+ return false;
+ }
+ }));
+ }
- assertEquals(numTasks, futures.size(),
- "Actual created task count did not match expected count");
+ assertEquals(numTasks, futures.size(),
+ "Actual created task count did not match expected count");
- int successCount = 0;
- for (Future<Boolean> future : futures) {
- if (future.get()) {
- successCount++;
+ int successCount = 0;
+ for (Future<Boolean> future : futures) {
+ if (future.get()) {
+ successCount++;
+ }
}
- }
- assertEquals(1, successCount, "Expected only one operation to succeed");
+ assertEquals(1, successCount, "Expected only one operation to
succeed");
- int tableCountAfter = client.tableOperations().list().size();
- assertTrue(client.tableOperations().exists(targetTableName),
- "Expected target table " + targetTableName + " to exist");
+ int tableCountAfter = client.tableOperations().list().size();
+ assertTrue(client.tableOperations().exists(targetTableName),
+ "Expected target table " + targetTableName + " to exist");
- String operation = successfulOperation.get();
- if ("create".equals(operation) || "clone".equals(operation)) {
- assertEquals(tableCountBefore + 1, tableCountAfter,
- "Expected +1 table count for " + operation);
- } else if ("rename".equals(operation)) {
- assertEquals(tableCountBefore, tableCountAfter, "Expected same table
count for rename");
+ String operation = successfulOperation.get();
+ if ("create".equals(operation) || "clone".equals(operation)) {
+ assertEquals(tableCountBefore + 1, tableCountAfter,
+ "Expected +1 table count for " + operation);
+ } else if ("rename".equals(operation)) {
+ assertEquals(tableCountBefore, tableCountAfter, "Expected same table
count for rename");
+ }
}
+ } finally {
+ pool.shutdown();
}
-
- pool.shutdown();
}
/**
@@ -298,30 +306,32 @@ public class ConcurrentTableNameOperationsIT extends
SharedMiniClusterBase {
final int numTasks = 16;
final int numIterations = 16;
ExecutorService pool = Executors.newFixedThreadPool(numTasks);
- String[] targetNamespaceNames = getUniqueNames(numIterations);
+ try {
+ String[] targetNamespaceNames = getUniqueNames(numIterations);
- for (String namespaceName : targetNamespaceNames) {
- Set<String> namespacesBefore = client.namespaceOperations().list();
+ for (String namespaceName : targetNamespaceNames) {
+ Set<String> namespacesBefore = client.namespaceOperations().list();
- int successCount = runConcurrentNamespaceOperation(pool, numTasks,
(index) -> {
- client.namespaceOperations().create(namespaceName);
- return true;
- });
+ int successCount = runConcurrentNamespaceOperation(pool, numTasks,
(index) -> {
+ client.namespaceOperations().create(namespaceName);
+ return true;
+ });
- assertEquals(1, successCount, "Expected only one create operation to
succeed");
- assertTrue(client.namespaceOperations().exists(namespaceName),
- "Expected namespace " + namespaceName + " to exist");
+ assertEquals(1, successCount, "Expected only one create operation to
succeed");
+ assertTrue(client.namespaceOperations().exists(namespaceName),
+ "Expected namespace " + namespaceName + " to exist");
- Set<String> namespacesAfter = client.namespaceOperations().list();
- Set<String> newNamespaces = new HashSet<>(namespacesAfter);
- newNamespaces.removeAll(namespacesBefore);
- assertEquals(Set.of(namespaceName), newNamespaces,
- "Expected exactly one new namespace: " + namespaceName);
+ Set<String> namespacesAfter = client.namespaceOperations().list();
+ Set<String> newNamespaces = new HashSet<>(namespacesAfter);
+ newNamespaces.removeAll(namespacesBefore);
+ assertEquals(Set.of(namespaceName), newNamespaces,
+ "Expected exactly one new namespace: " + namespaceName);
- client.namespaceOperations().delete(namespaceName);
+ client.namespaceOperations().delete(namespaceName);
+ }
+ } finally {
+ pool.shutdown();
}
-
- pool.shutdown();
}
/**
@@ -333,45 +343,47 @@ public class ConcurrentTableNameOperationsIT extends
SharedMiniClusterBase {
final int numTasks = 16;
final int numIterations = 8;
ExecutorService pool = Executors.newFixedThreadPool(numTasks);
- String[] targetNamespaceNames = getUniqueNames(numIterations);
-
- for (String targetNamespaceName : targetNamespaceNames) {
- // multiple source namespaces for rename ops
- List<String> sourceNamespaces = new ArrayList<>();
- for (int i = 0; i < numTasks; i++) {
- String sourceNamespace = targetNamespaceName + "_source_" + i;
- client.namespaceOperations().create(sourceNamespace);
- sourceNamespaces.add(sourceNamespace);
- }
+ try {
+ String[] targetNamespaceNames = getUniqueNames(numIterations);
+
+ for (String targetNamespaceName : targetNamespaceNames) {
+ // multiple source namespaces for rename ops
+ List<String> sourceNamespaces = new ArrayList<>();
+ for (int i = 0; i < numTasks; i++) {
+ String sourceNamespace = targetNamespaceName + "_source_" + i;
+ client.namespaceOperations().create(sourceNamespace);
+ sourceNamespaces.add(sourceNamespace);
+ }
- Set<String> namespacesBefore = client.namespaceOperations().list();
+ Set<String> namespacesBefore = client.namespaceOperations().list();
- int successCount = runConcurrentNamespaceOperation(pool, numTasks,
(index) -> {
- client.namespaceOperations().rename(sourceNamespaces.get(index),
targetNamespaceName);
- return true;
- });
+ int successCount = runConcurrentNamespaceOperation(pool, numTasks,
(index) -> {
+ client.namespaceOperations().rename(sourceNamespaces.get(index),
targetNamespaceName);
+ return true;
+ });
- assertEquals(1, successCount, "Expected only one rename operation to
succeed");
- assertTrue(client.namespaceOperations().exists(targetNamespaceName),
- "Expected target namespace " + targetNamespaceName + " to exist");
+ assertEquals(1, successCount, "Expected only one rename operation to
succeed");
+ assertTrue(client.namespaceOperations().exists(targetNamespaceName),
+ "Expected target namespace " + targetNamespaceName + " to exist");
- Set<String> namespacesAfter = client.namespaceOperations().list();
- assertEquals(namespacesBefore.size(), namespacesAfter.size(),
- "Expected same namespace count (rename operation)");
- assertTrue(namespacesAfter.contains(targetNamespaceName),
- "Expected target namespace in final list");
+ Set<String> namespacesAfter = client.namespaceOperations().list();
+ assertEquals(namespacesBefore.size(), namespacesAfter.size(),
+ "Expected same namespace count (rename operation)");
+ assertTrue(namespacesAfter.contains(targetNamespaceName),
+ "Expected target namespace in final list");
- for (String sourceNamespace : sourceNamespaces) {
- if (client.namespaceOperations().exists(sourceNamespace)) {
- client.namespaceOperations().delete(sourceNamespace);
+ for (String sourceNamespace : sourceNamespaces) {
+ if (client.namespaceOperations().exists(sourceNamespace)) {
+ client.namespaceOperations().delete(sourceNamespace);
+ }
+ }
+ if (client.namespaceOperations().exists(targetNamespaceName)) {
+ client.namespaceOperations().delete(targetNamespaceName);
}
}
- if (client.namespaceOperations().exists(targetNamespaceName)) {
- client.namespaceOperations().delete(targetNamespaceName);
- }
+ } finally {
+ pool.shutdown();
}
-
- pool.shutdown();
}
private int runConcurrentTableOperation(ExecutorService pool, int numTasks,
diff --git
a/test/src/main/java/org/apache/accumulo/test/functional/FateStarvationIT.java
b/test/src/main/java/org/apache/accumulo/test/functional/FateStarvationIT.java
index d80be33af9..de043de912 100644
---
a/test/src/main/java/org/apache/accumulo/test/functional/FateStarvationIT.java
+++
b/test/src/main/java/org/apache/accumulo/test/functional/FateStarvationIT.java
@@ -89,31 +89,35 @@ public class FateStarvationIT extends
AccumuloClusterHarness {
int numTasks = 100;
List<Future<?>> futures = new ArrayList<>(numTasks);
var executor = Executors.newCachedThreadPool();
- // wait for a portion of the tasks to be ready
- CountDownLatch startLatch = new CountDownLatch(32);
- assertTrue(numTasks >= startLatch.getCount(),
- "Not enough tasks to satisfy latch count - deadlock risk");
-
- for (int i = 0; i < numTasks; i++) {
- int idx1 = RANDOM.get().nextInt(splits.size() - 1);
- int idx2 = RANDOM.get().nextInt(splits.size() - (idx1 + 1)) + idx1 + 1;
-
- var future = executor.submit(() -> {
- startLatch.countDown();
- startLatch.await();
- c.tableOperations().compact(tableName, splits.get(idx1),
splits.get(idx2), false, true);
- return null;
- });
-
- futures.add(future);
- }
- assertEquals(numTasks, futures.size());
-
- log.debug("Started compactions");
-
- // wait for all compactions to complete
- for (var future : futures) {
- future.get();
+ try {
+ // wait for a portion of the tasks to be ready
+ CountDownLatch startLatch = new CountDownLatch(32);
+ assertTrue(numTasks >= startLatch.getCount(),
+ "Not enough tasks to satisfy latch count - deadlock risk");
+
+ for (int i = 0; i < numTasks; i++) {
+ int idx1 = RANDOM.get().nextInt(splits.size() - 1);
+ int idx2 = RANDOM.get().nextInt(splits.size() - (idx1 + 1)) + idx1 +
1;
+
+ var future = executor.submit(() -> {
+ startLatch.countDown();
+ startLatch.await();
+ c.tableOperations().compact(tableName, splits.get(idx1),
splits.get(idx2), false, true);
+ return null;
+ });
+
+ futures.add(future);
+ }
+ assertEquals(numTasks, futures.size());
+
+ log.debug("Started compactions");
+
+ // wait for all compactions to complete
+ for (var future : futures) {
+ future.get();
+ }
+ } finally {
+ executor.shutdown();
}
FunctionalTestUtils.assertNoDanglingFateLocks(getCluster());
diff --git
a/test/src/main/java/org/apache/accumulo/test/functional/FunctionalTestUtils.java
b/test/src/main/java/org/apache/accumulo/test/functional/FunctionalTestUtils.java
index cf2460035f..1b94f9116a 100644
---
a/test/src/main/java/org/apache/accumulo/test/functional/FunctionalTestUtils.java
+++
b/test/src/main/java/org/apache/accumulo/test/functional/FunctionalTestUtils.java
@@ -155,27 +155,30 @@ public class FunctionalTestUtils {
public static void createRFiles(final AccumuloClient c, final FileSystem fs,
String path,
int rows, int splits, int threads) throws Exception {
fs.delete(new Path(path), true);
- ExecutorService threadPool = Executors.newFixedThreadPool(threads);
final AtomicBoolean fail = new AtomicBoolean(false);
- for (int i = 0; i < rows; i += rows / splits) {
- TestIngest.IngestParams params = new
TestIngest.IngestParams(c.properties());
- params.outputFile = String.format("%s/mf%s", path, i);
- params.random = 56;
- params.timestamp = 1;
- params.dataSize = 50;
- params.rows = rows / splits;
- params.startRow = i;
- params.cols = 1;
- threadPool.execute(() -> {
- try {
- TestIngest.ingest(c, fs, params);
- } catch (Exception e) {
- fail.set(true);
- }
- });
+ ExecutorService threadPool = Executors.newFixedThreadPool(threads);
+ try {
+ for (int i = 0; i < rows; i += rows / splits) {
+ TestIngest.IngestParams params = new
TestIngest.IngestParams(c.properties());
+ params.outputFile = String.format("%s/mf%s", path, i);
+ params.random = 56;
+ params.timestamp = 1;
+ params.dataSize = 50;
+ params.rows = rows / splits;
+ params.startRow = i;
+ params.cols = 1;
+ threadPool.execute(() -> {
+ try {
+ TestIngest.ingest(c, fs, params);
+ } catch (Exception e) {
+ fail.set(true);
+ }
+ });
+ }
+ } finally {
+ threadPool.shutdown();
+ threadPool.awaitTermination(1, TimeUnit.HOURS);
}
- threadPool.shutdown();
- threadPool.awaitTermination(1, TimeUnit.HOURS);
assertFalse(fail.get());
}
diff --git
a/test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java
b/test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java
index b28ef8d3d9..aa660d7ed1 100644
---
a/test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java
+++
b/test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java
@@ -510,44 +510,48 @@ public class ManagerAssignmentIT extends
SharedMiniClusterBase {
};
ExecutorService service = Executors.newFixedThreadPool(10);
- for (int i = 0; i < 10; i++) {
- service.execute(task);
- }
-
- // Wait until all threads are reading some data
- latch.await();
-
- // getClusterControl().stopAllServers(ServerType.TABLET_SERVER)
- // could potentially send a kill -9 to the process. Shut the tablet
- // servers down in a more graceful way.
- final Map<String,Map<KeyExtent,List<Range>>> binnedRanges = new
HashMap<>();
- ((ClientContext)
client).getTabletLocationCache(tid).binRanges((ClientContext) client,
- Collections.singletonList(TabletsSection.getRange()), binnedRanges);
- binnedRanges.keySet().forEach((location) -> {
- HostAndPort address = HostAndPort.fromString(location);
- String addressWithSession = address.toString();
- var zLockPath = getCluster().getServerContext().getServerPaths()
- .createTabletServerPath(ResourceGroupId.DEFAULT, address);
- long sessionId =
-
ServiceLock.getSessionId(getCluster().getServerContext().getZooCache(),
zLockPath);
- if (sessionId != 0) {
- addressWithSession = address + "[" + Long.toHexString(sessionId) + "]";
+ try {
+ for (int i = 0; i < 10; i++) {
+ service.execute(task);
}
- final String finalAddress = addressWithSession;
- System.out.println("Attempting to shutdown TabletServer at: " + address);
- try {
- ThriftClientTypes.MANAGER.executeVoid((ClientContext) client,
- c -> c.shutdownTabletServer(TraceUtil.traceInfo(),
- getCluster().getServerContext().rpcCreds(), finalAddress,
false));
- } catch (AccumuloException | AccumuloSecurityException e) {
- fail("Error shutting down TabletServer", e);
- }
+ // Wait until all threads are reading some data
+ latch.await();
+
+ // getClusterControl().stopAllServers(ServerType.TABLET_SERVER)
+ // could potentially send a kill -9 to the process. Shut the tablet
+ // servers down in a more graceful way.
+ final Map<String,Map<KeyExtent,List<Range>>> binnedRanges = new
HashMap<>();
+ ((ClientContext)
client).getTabletLocationCache(tid).binRanges((ClientContext) client,
+ Collections.singletonList(TabletsSection.getRange()), binnedRanges);
+ binnedRanges.keySet().forEach((location) -> {
+ HostAndPort address = HostAndPort.fromString(location);
+ String addressWithSession = address.toString();
+ var zLockPath = getCluster().getServerContext().getServerPaths()
+ .createTabletServerPath(ResourceGroupId.DEFAULT, address);
+ long sessionId =
+
ServiceLock.getSessionId(getCluster().getServerContext().getZooCache(),
zLockPath);
+ if (sessionId != 0) {
+ addressWithSession = address + "[" + Long.toHexString(sessionId) +
"]";
+ }
- });
+ final String finalAddress = addressWithSession;
+ System.out.println("Attempting to shutdown TabletServer at: " +
address);
+ try {
+ ThriftClientTypes.MANAGER.executeVoid((ClientContext) client,
+ c -> c.shutdownTabletServer(TraceUtil.traceInfo(),
+ getCluster().getServerContext().rpcCreds(), finalAddress,
false));
+ } catch (AccumuloException | AccumuloSecurityException e) {
+ fail("Error shutting down TabletServer", e);
+ }
- Wait.waitFor(
- () ->
client.instanceOperations().getServers(ServerId.Type.TABLET_SERVER).size() ==
0);
+ });
+
+ Wait.waitFor(
+ () ->
client.instanceOperations().getServers(ServerId.Type.TABLET_SERVER).size() ==
0);
+ } finally {
+ service.shutdownNow();
+ }
// restart the tablet server for the other tests. Need to call
stopAllServers
// to clear out the process list because we shutdown the TabletServer
outside
diff --git
a/test/src/main/java/org/apache/accumulo/test/functional/ScanIdIT.java
b/test/src/main/java/org/apache/accumulo/test/functional/ScanIdIT.java
index c9c0d33d22..837a1202b9 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/ScanIdIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/ScanIdIT.java
@@ -63,6 +63,8 @@ import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.ColumnVisibility;
import org.apache.accumulo.test.harness.AccumuloClusterHarness;
import org.apache.hadoop.io.Text;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -93,10 +95,20 @@ public class ScanIdIT extends AccumuloClusterHarness {
private static final int NUM_BATCH_SCANNERS = 1;
private static final int NUM_TOTAL_SCANNERS = NUM_SINGLE_SCANNERS +
NUM_BATCH_SCANNERS;
private static final int NUM_DATA_ROWS = 100;
- private static final ExecutorService pool =
Executors.newFixedThreadPool(NUM_TOTAL_SCANNERS);
+ private static ExecutorService pool;
private static final AtomicBoolean testInProgress = new AtomicBoolean(true);
private static final Map<Integer,Value> resultsByWorker = new
ConcurrentHashMap<>();
+ @BeforeAll
+ public static void setup() {
+ pool = Executors.newFixedThreadPool(NUM_TOTAL_SCANNERS);
+ }
+
+ @AfterAll
+ public static void teardown() {
+ pool.shutdownNow();
+ }
+
@Override
protected Duration defaultTimeout() {
return Duration.ofMinutes(2);
diff --git
a/test/src/main/java/org/apache/accumulo/test/functional/SplitIT.java
b/test/src/main/java/org/apache/accumulo/test/functional/SplitIT.java
index a63050583d..aba4391e2a 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/SplitIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/SplitIT.java
@@ -385,68 +385,69 @@ public class SplitIT extends AccumuloClusterHarness {
log.debug("Creating futures that add random splits to the table");
ExecutorService es = Executors.newFixedThreadPool(10);
- final int totalFutures = 100;
- final int splitsPerFuture = 4;
- final Set<Text> totalSplits = new ConcurrentSkipListSet<>();
- List<Callable<Void>> tasks = new ArrayList<>(totalFutures);
- for (int i = 0; i < totalFutures; i++) {
- final Pair<Integer,Integer> splitBounds =
getRandomSplitBounds(numRows);
- final TreeSet<Text> splits =
TestIngest.getSplitPoints(splitBounds.getFirst().longValue(),
- splitBounds.getSecond().longValue(), splitsPerFuture);
- tasks.add(() -> {
- c.tableOperations().addSplits(tableName, splits);
- totalSplits.addAll(splits);
- return null;
- });
- }
+ try {
+ final int totalFutures = 100;
+ final int splitsPerFuture = 4;
+ final Set<Text> totalSplits = new ConcurrentSkipListSet<>();
+ List<Callable<Void>> tasks = new ArrayList<>(totalFutures);
+ for (int i = 0; i < totalFutures; i++) {
+ final Pair<Integer,Integer> splitBounds =
getRandomSplitBounds(numRows);
+ final TreeSet<Text> splits =
TestIngest.getSplitPoints(splitBounds.getFirst().longValue(),
+ splitBounds.getSecond().longValue(), splitsPerFuture);
+ tasks.add(() -> {
+ c.tableOperations().addSplits(tableName, splits);
+ totalSplits.addAll(splits);
+ return null;
+ });
+ }
- log.debug("Submitting futures");
- List<Future<Void>> futures =
- tasks.parallelStream().map(es::submit).collect(Collectors.toList());
+ log.debug("Submitting futures");
+ List<Future<Void>> futures =
+
tasks.parallelStream().map(es::submit).collect(Collectors.toList());
- Set<Text> splitsAfterOffline = null;
- if (offlineTable) {
- // run offline concurrently with split operation
- c.tableOperations().offline(tableName, true);
- splitsAfterOffline =
Set.copyOf(c.tableOperations().listSplits(tableName));
- }
+ Set<Text> splitsAfterOffline = null;
+ if (offlineTable) {
+ // run offline concurrently with split operation
+ c.tableOperations().offline(tableName, true);
+ splitsAfterOffline =
Set.copyOf(c.tableOperations().listSplits(tableName));
+ }
- log.debug("Waiting for futures to complete");
- for (Future<?> f : futures) {
- try {
- f.get();
- } catch (ExecutionException ee) {
- if (offlineTable && ee.getMessage().contains("is offline")) {
- // Some exceptions are expected when concurrently taking the table
offline.
- log.debug(ee.getMessage());
- } else {
- throw ee;
+ log.debug("Waiting for futures to complete");
+ for (Future<?> f : futures) {
+ try {
+ f.get();
+ } catch (ExecutionException ee) {
+ if (offlineTable && ee.getMessage().contains("is offline")) {
+ // Some exceptions are expected when concurrently taking the
table offline.
+ log.debug(ee.getMessage());
+ } else {
+ throw ee;
+ }
}
}
- }
-
- if (offlineTable) {
- // The splits seen immediately after offline() call should not change
after all the futures
- // complete. This ensures that nothing changes in the tablet after the
offline+wait call
- // returns.
- assertEquals(splitsAfterOffline, new
HashSet<>(c.tableOperations().listSplits(tableName)),
- "Splits changed after offline");
-
- // table will be scanned for verification, so bring it online
- c.tableOperations().online(tableName);
- } else {
- assertFalse(totalSplits.isEmpty());
- }
-
- log.debug("Checking that {} splits were created ", totalSplits.size());
- assertEquals(totalSplits, new
HashSet<>(c.tableOperations().listSplits(tableName)),
- "Did not see expected splits");
- log.debug("Verifying {} rows ingested into {}", numRows, tableName);
- VerifyIngest.verifyIngest(c, params);
+ if (offlineTable) {
+ // The splits seen immediately after offline() call should not
change after all the
+ // futures complete. This ensures that nothing changes in the tablet
after the
+ // offline+wait call returns.
+ assertEquals(splitsAfterOffline, new
HashSet<>(c.tableOperations().listSplits(tableName)),
+ "Splits changed after offline");
+
+ // table will be scanned for verification, so bring it online
+ c.tableOperations().online(tableName);
+ } else {
+ assertFalse(totalSplits.isEmpty());
+ }
- es.shutdown();
+ log.debug("Checking that {} splits were created ", totalSplits.size());
+ assertEquals(totalSplits, new
HashSet<>(c.tableOperations().listSplits(tableName)),
+ "Did not see expected splits");
+ log.debug("Verifying {} rows ingested into {}", numRows, tableName);
+ VerifyIngest.verifyIngest(c, params);
+ } finally {
+ es.shutdown();
+ }
}
}
diff --git
a/test/src/main/java/org/apache/accumulo/test/functional/WriteLotsIT.java
b/test/src/main/java/org/apache/accumulo/test/functional/WriteLotsIT.java
index de8fe05aa8..ce75154c72 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/WriteLotsIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/WriteLotsIT.java
@@ -54,21 +54,24 @@ public class WriteLotsIT extends AccumuloClusterHarness {
final int THREADS = 5;
ThreadPoolExecutor tpe = new ThreadPoolExecutor(0, THREADS, 0,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(THREADS));
- for (int i = 0; i < THREADS; i++) {
- final int index = i;
- Runnable r = () -> {
- try {
- IngestParams ingestParams = new IngestParams(getClientProps(),
tableName, 10_000);
- ingestParams.startRow = index * 10000;
- TestIngest.ingest(c, ingestParams);
- } catch (Exception ex) {
- ref.set(ex);
- }
- };
- tpe.execute(r);
+ try {
+ for (int i = 0; i < THREADS; i++) {
+ final int index = i;
+ Runnable r = () -> {
+ try {
+ IngestParams ingestParams = new IngestParams(getClientProps(),
tableName, 10_000);
+ ingestParams.startRow = index * 10000;
+ TestIngest.ingest(c, ingestParams);
+ } catch (Exception ex) {
+ ref.set(ex);
+ }
+ };
+ tpe.execute(r);
+ }
+ } finally {
+ tpe.shutdown();
+ tpe.awaitTermination(90, TimeUnit.SECONDS);
}
- tpe.shutdown();
- tpe.awaitTermination(90, TimeUnit.SECONDS);
if (ref.get() != null) {
throw ref.get();
}
