Skip to content

Properly close thread pools #6480

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
DomGarguilo merged 3 commits into from
Jul 20, 2026
Merged

Properly close thread pools #6480

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
410ca1a
Triage Executors.new* in code
DomGarguilo Jul 16, 2026
07f290b
comment formatting
DomGarguilo Jul 16, 2026
4d88a77
Address one more spot
DomGarguilo Jul 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
190 changes: 99 additions & 91 deletions core/src/test/java/org/apache/accumulo/core/clientImpl/ClientTabletCacheImplTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Future<CachedTablet>> futures = new ArrayList<>(numTasks);
CountDownLatch startLatch = new CountDownLatch(32); // start a portion of threads at once
assertTrue(numTasks >= startLatch.getCount(),
"Not enough tasks to satisfy latch count - deadlock risk");

// multiple rounds to increase the chance of contention
for (int round = 0; round < roundCount; round++) {
try {
final int lookupCount = 128;
final int roundCount = 8;
final int numTasks = roundCount * lookupCount;

List<Future<CachedTablet>> futures = new ArrayList<>(numTasks);
CountDownLatch startLatch = new CountDownLatch(32); // start a portion of threads at once
assertTrue(numTasks >= startLatch.getCount(),
"Not enough tasks to satisfy latch count - deadlock risk");

// multiple rounds to increase the chance of contention
for (int round = 0; round < roundCount; round++) {

// start a bunch of threads all trying to lookup data in the cache
// should see exactly 2 threads doing metadata lookups at a time
List<String> rowsToLookup = new ArrayList<>(lookupCount);

for (int i = 0; i < lookupCount; i++) {
String lookup = (char) ('a' + (i % 26)) + "";
rowsToLookup.add(lookup);
}

// start a bunch of threads all trying to lookup data in the cache
// should see exactly 2 threads doing metadata lookups at a time
List<String> rowsToLookup = new ArrayList<>(lookupCount);
Collections.shuffle(rowsToLookup, RANDOM.get());

for (int i = 0; i < lookupCount; i++) {
String lookup = (char) ('a' + (i % 26)) + "";
rowsToLookup.add(lookup);
for (var lookup : rowsToLookup) {
var future = executor.submit(() -> {
startLatch.countDown();
startLatch.await();
if (RANDOM.get().nextInt(10) < 3) {
Thread.yield();
}
var loc = metaCache.findTablet(context, new Text(lookup), false, LocationNeed.REQUIRED);
if (lookup.compareTo("m") <= 0) {
assertEquals("tserver7", loc.getTserverLocation().orElseThrow());
} else if (lookup.compareTo("q") <= 0) {
assertEquals("tserver8", loc.getTserverLocation().orElseThrow());
} else {
assertEquals("tserver9", loc.getTserverLocation().orElseThrow());
}
return loc;
});
futures.add(future);
}
}
assertEquals(numTasks, futures.size());

Collections.shuffle(rowsToLookup, RANDOM.get());

for (var lookup : rowsToLookup) {
var future = executor.submit(() -> {
startLatch.countDown();
startLatch.await();
if (RANDOM.get().nextInt(10) < 3) {
Thread.yield();
}
var loc = metaCache.findTablet(context, new Text(lookup), false, LocationNeed.REQUIRED);
if (lookup.compareTo("m") <= 0) {
assertEquals("tserver7", loc.getTserverLocation().orElseThrow());
} else if (lookup.compareTo("q") <= 0) {
assertEquals("tserver8", loc.getTserverLocation().orElseThrow());
} else {
assertEquals("tserver9", loc.getTserverLocation().orElseThrow());
}
return loc;
});
futures.add(future);
for (var future : futures) {
assertNotNull(future.get());
}
}
assertEquals(numTasks, futures.size());

for (var future : futures) {
assertNotNull(future.get());
assertTrue(sawTwoActive.get(), "Expected to see exactly two lookups.");
// The second metadata tablet (mte2) contains two user tablets (ke2 and ke3). Depending on
// which of these two user tablets is looked up in the metadata table first will see a total
// of 2 or 3 lookups. If the location of ke2 is looked up first then it will get the locations
// of ke2 and ke3 from mte2 and put them in the cache. If the location of ke3 is looked up
// first then it will only get the location of ke3 from mte2 and not ke2.
assertTrue(lookups.size() == 2 || lookups.size() == 3,
"Expected 2 or 3 lookups, got " + lookups.size() + " : " + lookups);
assertEquals(1,
lookups.stream().filter(metadataExtent -> metadataExtent.equals(mte1)).count(),
lookups::toString);
var mte2Lookups =
lookups.stream().filter(metadataExtent -> metadataExtent.equals(mte2)).count();
assertTrue(mte2Lookups == 1 || mte2Lookups == 2, lookups::toString);
} finally {
executor.shutdown();
}

assertTrue(sawTwoActive.get(), "Expected to see exactly two lookups.");
// The second metadata tablet (mte2) contains two user tablets (ke2 and ke3). Depending on which
// of these two user tablets is looked up in the metadata table first will see a total of 2 or 3
// lookups. If the location of ke2 is looked up first then it will get the locations of ke2 and
// ke3 from mte2 and put them in the cache. If the location of ke3 is looked up first then it
// will only get the location of ke3 from mte2 and not ke2.
assertTrue(lookups.size() == 2 || lookups.size() == 3,
"Expected 2 or 3 lookups, got " + lookups.size() + " : " + lookups);
assertEquals(1, lookups.stream().filter(metadataExtent -> metadataExtent.equals(mte1)).count(),
lookups::toString);
var mte2Lookups =
lookups.stream().filter(metadataExtent -> metadataExtent.equals(mte2)).count();
assertTrue(mte2Lookups == 1 || mte2Lookups == 2, lookups::toString);
executor.shutdown();
}

@Test
Expand Down Expand Up @@ -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
Expand Down
58 changes: 31 additions & 27 deletions core/src/test/java/org/apache/accumulo/core/crypto/CryptoTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Future<Boolean>> verifyFutures = new ArrayList<>(numTasks);
CountDownLatch startLatch = new CountDownLatch(numTasks);
assertTrue(numTasks >= startLatch.getCount(),
"Not enough tasks to satisfy latch count - deadlock risk");

FileDecrypter decrypter = cs.getFileDecrypter(new CryptoEnvironmentImpl(scope, null, params));

// verify that each input stream returned by decrypter.decryptStream() is independent when
// used by multiple threads
for (int i = 0; i < numTasks; i++) {
var future = executor.submit(() -> {
startLatch.countDown();
startLatch.await();
try (ByteArrayInputStream in = new ByteArrayInputStream(cipherText);
DataInputStream decrypted = new DataInputStream(decrypter.decryptStream(in))) {
byte[] dataRead = new byte[plainText.length];
decrypted.readFully(dataRead);
return Arrays.equals(plainText, dataRead);
}
});
verifyFutures.add(future);
}
assertEquals(numTasks, verifyFutures.size());

final int numTasks = 32;
List<Future<Boolean>> verifyFutures = new ArrayList<>(numTasks);
CountDownLatch startLatch = new CountDownLatch(numTasks);
assertTrue(numTasks >= startLatch.getCount(),
"Not enough tasks to satisfy latch count - deadlock risk");

FileDecrypter decrypter = cs.getFileDecrypter(new CryptoEnvironmentImpl(scope, null, params));

// verify that each input stream returned by decrypter.decryptStream() is independent when used
// by multiple threads
for (int i = 0; i < numTasks; i++) {
var future = executor.submit(() -> {
startLatch.countDown();
startLatch.await();
try (ByteArrayInputStream in = new ByteArrayInputStream(cipherText);
DataInputStream decrypted = new DataInputStream(decrypter.decryptStream(in))) {
byte[] dataRead = new byte[plainText.length];
decrypted.readFully(dataRead);
return Arrays.equals(plainText, dataRead);
}
});
verifyFutures.add(future);
}
assertEquals(numTasks, verifyFutures.size());

for (var future : verifyFutures) {
assertTrue(future.get());
for (var future : verifyFutures) {
assertTrue(future.get());
}
} finally {
executor.shutdownNow();
}
}

Expand Down
Loading