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 a2cf16c176b..9b1cb1adecc 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 void testMultithreadedLookups() throws Exception { setLocation(tservers, "tserver3", mte2, ke3, "tserver9"); var executor = Executors.newCachedThreadPool(); - final int lookupCount = 128; - final int roundCount = 8; - final int numTasks = roundCount * lookupCount; - - List> 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> 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 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 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 void testNonBlocking() throws Exception { 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 b7833bedebf..668df2a26bf 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 @@ private void testMultipleThreads(Scope scope) throws Exception { byte[] cipherText = out.toByteArray(); var executor = Executors.newCachedThreadPool(); + try { + + final int numTasks = 32; + List> 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> 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 ccbb1f0974e..c4c2f9ca306 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 void testManyStartNotNull() throws InterruptedException, ExecutionExcepti final int numTasks = 32; ExecutorService service = Executors.newFixedThreadPool(numTasks); - - ArrayList> 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 result : results) { - assertTrue(result.get(), - al + " resulted in a failed call to getcodec within the thread pool"); + try { + + ArrayList> 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 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 void testManyDontStartUntilThread() throws InterruptedException, Executio final int numTasks = 32; ExecutorService service = Executors.newFixedThreadPool(numTasks); - - ArrayList> 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 result : results) { - assertTrue(result.get(), - al + " resulted in a failed call to getcodec within the thread pool"); + try { + + ArrayList> 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 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 void testThereCanBeOnlyOne() throws InterruptedException, ExecutionExcept final int numTasks = 32; ExecutorService service = Executors.newFixedThreadPool(numTasks); - - ArrayList> 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 hashCodes = new HashSet<>(); - for (Future result : service.invokeAll(list)) { - hashCodes.add(result.get()); + try { + + ArrayList> 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 hashCodes = new HashSet<>(); + for (Future 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 2e719540b67..deaa61a0146 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 void testMerge() throws Exception { @Test public void testIterateUntil() throws Exception { ExecutorService es = Executors.newFixedThreadPool(1); - Function> step = - n -> CompletableFuture.supplyAsync(() -> n - 1, es); - Predicate 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> 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> 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 badIsDone = n -> { - throw new RuntimeException(); - }; - assertThrows(ExecutionException.class, - () -> CompletableFutureUtil.iterateUntil(step, badIsDone, 100).get()); + try { + Function> step = + n -> CompletableFuture.supplyAsync(() -> n - 1, es); + Predicate 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> 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> 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 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 6120cf63109..6707061be81 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 @@ private void testRange(long start, long end) { 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> 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> 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 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 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 d34cc2666d3..016a784704c 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 void testAddPollRaceCondition() throws Exception { Stream.of("G1", "G2", "G3").map(ResourceGroupId::of).toArray(ResourceGroupId[]::new); var executor = Executors.newFixedThreadPool(groups.length); + try { - List> futures = new ArrayList<>(); + List> 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 4a90cca43f9..c0bdc1d1d19 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 8deb58a6a41..4ca08358688 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 void compareOldNewBulkImportTest() throws Exception { fs.mkdirs(base); ExecutorService es = Executors.newFixedThreadPool(5); - List> 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> 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 dirs = new ArrayList<>(); + for (Future f : futures) { + dirs.add(f.get()); + } + log.info("Importing"); + long startOps = getStat(getStats(), "FileInfoOps"); + long now = System.currentTimeMillis(); + List> errs = new ArrayList<>(); + for (String dir : dirs) { + errs.add(es.submit(() -> { + c.tableOperations().importDirectory(dir).to(tableName).load(); + return null; + })); + } + for (Future 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 dirs = new ArrayList<>(); - for (Future f : futures) { - dirs.add(f.get()); - } - log.info("Importing"); - long startOps = getStat(getStats(), "FileInfoOps"); - long now = System.currentTimeMillis(); - List> errs = new ArrayList<>(); - for (String dir : dirs) { - errs.add(es.submit(() -> { - c.tableOperations().importDirectory(dir).to(tableName).load(); - return null; - })); - } - for (Future 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 b3f0cdc305b..750da22d3cf 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 @@ private void testLargeMemory(NewTableConfiguration config, Consumer sca 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 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 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 dce79c99535..a15b1777f61 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 void testFate() throws Exception { // 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 0c3140c38dc..e82299c5764 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 void testScanTabletsWithOperationIds() throws Exception { 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 62d67a99faa..b92556991be 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 void testMaxLatency() throws Exception { // 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 27899f38d9c..23fd5a49080 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 void test() throws Exception { ExecutorService svc = Executors.newFixedThreadPool(numThreads); CountDownLatch countDown = new CountDownLatch(numThreads); ArrayList> 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 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 21c835b0d36..610b2b74498 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 void testDefendAgainstThreadsCreateSameTableNameConcurrently() 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> futureList = new ArrayList<>(numTasks); - - for (int j = 0; j < numTasks; j++) { - Future 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> futureList = new ArrayList<>(numTasks); + + for (int j = 0; j < numTasks; j++) { + Future 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 result : futureList) { - if (result.get() == true) { - taskSucceeded++; - } else { - taskFailed++; + int taskSucceeded = 0; + int taskFailed = 0; + for (Future 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 void testUniquenessOfTableId() throws ExecutionException, InterruptedExce Set 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 future = pool.submit(() -> { - TableId tableId = null; + Future 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 tab : futureList) { - hash.add(tab.get()); + for (Future 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 facbe5e1b5d..5d8087b831e 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 @@ void testConcurrentNameGeneration() throws Exception { Set namesSeen = ConcurrentHashMap.newKeySet(); var executorService = Executors.newCachedThreadPool(); - final int numLargeTasks = 64; - final int numSmallTasks = 10; - final int numTasks = numLargeTasks + numSmallTasks; - List> 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> 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 80559f24152..60283f74231 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 void testZombieScan() throws Exception { String table = getUniqueNames(1)[0]; + var executor = Executors.newCachedThreadPool(); try (AccumuloClient c = Accumulo.newClient().from(getClientProperties()).build()) { var splits = new TreeSet(); @@ -170,8 +171,6 @@ public void testZombieScan() throws Exception { // 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> futures = new ArrayList<>(); for (var row : List.of("2", "4")) { @@ -252,6 +251,7 @@ public void testZombieScan() throws Exception { // simple enough to include assertValidScanIds(c); + } finally { executor.shutdownNow(); } @@ -273,12 +273,11 @@ public void testMetrics(ConsistencyLevel consistency) throws Exception { 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> futures = new ArrayList<>(); for (var row : List.of("2", "4")) { @@ -337,6 +336,7 @@ public void testMetrics(ConsistencyLevel consistency) throws Exception { 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 3039cbb3ed8..c1bdbdbe9a1 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 Map getProperties() throws Exception { 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>(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>(numTasks); + + final int iterations = 151; + + tasks.add(() -> { + startLatch.countDown(); + startLatch.await(); + for (int i = 0; i < iterations; i++) { + + Map prevProps = null; + if (i % 10 == 0) { + prevProps = propShim.getProperties(); + } + + Map 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 prevProps = null; - if (i % 10 == 0) { - prevProps = propShim.getProperties(); + tableProps.put("general.custom.B", B + 17 + ""); + }); } + return null; + }); - Map 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 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 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 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 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 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 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 f0ad481e6b3..d689db45faf 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 void concurrentMutatorTest() throws Exception { 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>> futures = new ArrayList<>(numTasks); @@ -131,7 +130,6 @@ public void concurrentMutatorTest() throws Exception { 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 void concurrentMutatorTest() throws Exception { 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 52cd4550f83..b85679b1150 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 @@ private static Optional computeRemaining(TabletMergeabilityInfo tmi) { @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 void concurrentAddSplitTest() throws Exception { 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 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 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 36ed26a3fb3..474b8edc110 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 void testPause() throws Exception { // 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 void testManyTabletAndFiles() throws Exception { c.tableOperations().addSplits(tableName, splits); final int numTasks = 16; - var executor = Executors.newFixedThreadPool(numTasks); var futures = new ArrayList>(); // wait for a portion of the tasks to be ready CountDownLatch startLatch = new CountDownLatch(numTasks); @@ -1212,27 +1215,30 @@ public void testManyTabletAndFiles() throws Exception { 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 2e6fbfa7d93..000ffd26e07 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 void testSuccessfulCompaction() throws Exception { 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 void testOfflineAndCompactions() throws Exception { 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> 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 275223b1928..64de2ab6cd9 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 @@ protected Duration defaultTimeout() { @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 void testConcurrentDeleteTablesOps() throws Exception { 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 @@ protected void doTableOp() throws Exception { 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 571fbbd334f..ea5642ac5aa 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 void cloneTable() throws Exception { final int numTasks = 16; final int numIterations = 8; ExecutorService pool = Executors.newFixedThreadPool(numTasks); + try { + + for (String targetTableName : getUniqueNames(numIterations)) { + List 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 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 void renameTable() throws Exception { final int numTasks = 16; final int numIterations = 10; ExecutorService pool = Executors.newFixedThreadPool(numTasks); + try { + + for (String targetTableName : getUniqueNames(numIterations)) { + List 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 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 void importTable() throws Exception { 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 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 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 void mixedTableOperations() throws Exception { final int numTasks = operationsPerType * 3; final int numIterations = 4; ExecutorService pool = Executors.newFixedThreadPool(numTasks); - String[] expectedTableNames = getUniqueNames(numIterations); - - for (String targetTableName : expectedTableNames) { - List cloneSourceTables = new ArrayList<>(); - List 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 cloneSourceTables = new ArrayList<>(); + List 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> futures = new ArrayList<>(); - CountDownLatch startSignal = new CountDownLatch(numTasks); - AtomicReference 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> futures = new ArrayList<>(); + CountDownLatch startSignal = new CountDownLatch(numTasks); + AtomicReference 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 future : futures) { - if (future.get()) { - successCount++; + int successCount = 0; + for (Future 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 void createNamespace() throws Exception { 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 namespacesBefore = client.namespaceOperations().list(); + for (String namespaceName : targetNamespaceNames) { + Set 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 namespacesAfter = client.namespaceOperations().list(); - Set newNamespaces = new HashSet<>(namespacesAfter); - newNamespaces.removeAll(namespacesBefore); - assertEquals(Set.of(namespaceName), newNamespaces, - "Expected exactly one new namespace: " + namespaceName); + Set namespacesAfter = client.namespaceOperations().list(); + Set 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 void renameNamespace() throws Exception { 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 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 sourceNamespaces = new ArrayList<>(); + for (int i = 0; i < numTasks; i++) { + String sourceNamespace = targetNamespaceName + "_source_" + i; + client.namespaceOperations().create(sourceNamespace); + sourceNamespaces.add(sourceNamespace); + } - Set namespacesBefore = client.namespaceOperations().list(); + Set 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 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 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 d80be33af93..de043de9126 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 void run() throws Exception { int numTasks = 100; List> 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 cf2460035f8..1b94f9116a7 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 static void checkSplits(AccumuloClient c, String table, int min, int max) 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 b28ef8d3d9d..aa660d7ed10 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 void testShutdownOnlyTServerWithUserTable() throws Exception { }; 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>> 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>> 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 c9c0d33d225..837a1202b9b 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.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 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 a63050583d4..aba4391e2a4 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 void concurrentSplit(boolean offlineTable) throws Exception { 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 totalSplits = new ConcurrentSkipListSet<>(); - List> tasks = new ArrayList<>(totalFutures); - for (int i = 0; i < totalFutures; i++) { - final Pair splitBounds = getRandomSplitBounds(numRows); - final TreeSet 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 totalSplits = new ConcurrentSkipListSet<>(); + List> tasks = new ArrayList<>(totalFutures); + for (int i = 0; i < totalFutures; i++) { + final Pair splitBounds = getRandomSplitBounds(numRows); + final TreeSet 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> futures = - tasks.parallelStream().map(es::submit).collect(Collectors.toList()); + log.debug("Submitting futures"); + List> futures = + tasks.parallelStream().map(es::submit).collect(Collectors.toList()); - Set splitsAfterOffline = null; - if (offlineTable) { - // run offline concurrently with split operation - c.tableOperations().offline(tableName, true); - splitsAfterOffline = Set.copyOf(c.tableOperations().listSplits(tableName)); - } + Set 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 de8fe05aa88..ce75154c72f 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 void writeLots() throws Exception { 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(); }