Skip to content

Implement a closeAll utility to close multiple closeables #3470

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
merged 4 commits into from
Jul 14, 2025
Merged

Implement a closeAll utility to close multiple closeables #3470

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b545d25
Implement and use a "closeAll" utility
ohadzeliger Jul 7, 2025
64da875
Style
ohadzeliger Jul 7, 2025
a6c0305
PR comments: Move utility to CLoseableUtils.
ohadzeliger Jul 8, 2025
9211256
PR comments
ohadzeliger Jul 10, 2025
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
38 changes: 38 additions & 0 deletions fdb-extensions/src/main/java/com/apple/foundationdb/async/MoreAsyncUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -912,6 +912,32 @@ public static void closeIterator(@Nonnull Iterator<?> iterator) {
}
}

/**
* A utility to close multiple {@link AutoCloseable} objects, preserving all the caught exceptions.
* The method would attempt to close all closeables in order, even if some failed.
* @param closeables the given sequence of {@link AutoCloseable}
* @throws CloseException in case any exception was caught during the process. The first exception will be added
* as a {@code cause}. In case more than one exception was caught, it will be added as Suppressed.
*/
@SuppressWarnings("PMD.CloseResource")
public static void closeAll(AutoCloseable... closeables) throws CloseException {
CloseException accumulatedException = null;
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why create a new exception type to accumulate, rather than setting it to the first exception thrown?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That was how I initially tried to implement this but the compiler complained about "close() method can throw InterruptedException" when the close() declares to throw Exception.
Similar to the issue described here: TritonDataCenter/java-manta#322

for (AutoCloseable closeable: closeables) {
try {
closeable.close();
} catch (Exception e) {
if (accumulatedException == null) {
accumulatedException = new CloseException(e);
} else {
accumulatedException.addSuppressed(e);
}
}
}
if (accumulatedException != null) {
throw accumulatedException;
}
}

/**
* This is supposed to replicate the semantics of {@link java.util.concurrent.CompletionStage#whenComplete(BiConsumer)}
* but to handle the case where the completion handler might itself contain async work.
Expand Down Expand Up @@ -1081,4 +1107,16 @@ private DeadlineExceededException(long deadlineTimeMillis) {
addLogInfo("deadlineTimeMillis", deadlineTimeMillis);
}
}

/**
* Exception thrown when the {@link #closeAll} method catches an exception.
* This exception will have the cause set to the first exception thrown during {@code closeAll} and any further
* exception thrown will be added as {@Suppressed}.
*/
@SuppressWarnings("serial")
public static class CloseException extends Exception {
public CloseException(final Throwable cause) {
super(cause);
}
}
}
72 changes: 72 additions & 0 deletions fdb-extensions/src/test/java/com/apple/foundationdb/async/MoreAsyncUtilTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import com.apple.test.ParameterizedTestUtils;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.hamcrest.Matcher;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
Expand Down Expand Up @@ -273,8 +274,56 @@ void swallowException() throws ExecutionException, InterruptedException {
throw new RuntimeException();
}).get();
}
}

@Test
void closeAllNoIssue() throws Exception {
SimpleCloseable c1 = new SimpleCloseable(false, null);
SimpleCloseable c2 = new SimpleCloseable(false, null);
SimpleCloseable c3 = new SimpleCloseable(false, null);

MoreAsyncUtil.closeAll(c1, c2, c3);

Assertions.assertTrue(c1.isClosed());
Assertions.assertTrue(c2.isClosed());
Assertions.assertTrue(c3.isClosed());
}

@Test
void closeAllFailed() throws Exception {
SimpleCloseable c1 = new SimpleCloseable(true, "c1");
SimpleCloseable c2 = new SimpleCloseable(true, "c2");
SimpleCloseable c3 = new SimpleCloseable(true, "c3");

final MoreAsyncUtil.CloseException exception = assertThrows(MoreAsyncUtil.CloseException.class, () -> MoreAsyncUtil.closeAll(c1, c2, c3));

Assertions.assertEquals("c1", exception.getCause().getMessage());
final Throwable[] suppressed = exception.getSuppressed();
Assertions.assertEquals(2, suppressed.length);
Assertions.assertEquals("c2", suppressed[0].getMessage());
Assertions.assertEquals("c3", suppressed[1].getMessage());

Assertions.assertTrue(c1.isClosed());
Assertions.assertTrue(c2.isClosed());
Assertions.assertTrue(c3.isClosed());
}

@Test
void closeSomeFailed() throws Exception {
SimpleCloseable c1 = new SimpleCloseable(true, "c1");
SimpleCloseable c2 = new SimpleCloseable(false, null);
SimpleCloseable c3 = new SimpleCloseable(true, "c3");

final MoreAsyncUtil.CloseException exception = assertThrows(MoreAsyncUtil.CloseException.class, () -> MoreAsyncUtil.closeAll(c1, c2, c3));

Assertions.assertEquals("c1", exception.getCause().getMessage());
final Throwable[] suppressed = exception.getSuppressed();
Assertions.assertEquals(1, suppressed.length);
Assertions.assertEquals("c3", suppressed[0].getMessage());

Assertions.assertTrue(c1.isClosed());
Assertions.assertTrue(c2.isClosed());
Assertions.assertTrue(c3.isClosed());
}

private static void assertSwallowedOrNot(final CompletableFuture<Void> completedExceptionally1, final RuntimeException runtimeException1) throws InterruptedException, ExecutionException {
Expand All @@ -295,4 +344,27 @@ private static Matcher<String> isCurrentThreadNameOr(@Nonnull String threadName)
private static Matcher<String> isCurrentThreadNameOr(@Nonnull Matcher<String> threadMatcher) {
return either(threadMatcher).or(equalTo(Thread.currentThread().getName()));
}

private class SimpleCloseable implements AutoCloseable {
private final boolean fail;
private final String message;
private boolean closed = false;

public SimpleCloseable(boolean fail, String message) {
this.fail = fail;
this.message = message;
}

@Override
public void close() {
closed = true;
if (fail) {
throw new RuntimeException(message);
}
}

public boolean isClosed() {
return closed;
}
}
}
9 changes: 8 additions & 1 deletion ...n/java/com/apple/foundationdb/record/provider/foundationdb/recordrepair/RecordRepair.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,7 +115,14 @@ public static Builder builder(@Nonnull FDBDatabase database, final FDBRecordStor

@Override
public void close() {
throttledIterator.close();
try {
throttledIterator.close();
} catch (Exception e) {
if (logger.isWarnEnabled()) {
logger.warn("Failed to close the throttled iterator", e);
}
// Do not rethrow. We are trying to close the runner and the exception should log all errors
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why the change here?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I thought this would be a better fit to a usage like:

    RepairValidationResults repairResults;
    try (RecordRepairValidateRunner repairRunner = builder.buildRepairRunner(false)) {
        repairResults = repairRunner.run().join();
    }
    // Handle results

Where a failure to close would disrupt the flow and cause the loss of already collected results.
But I am not too strongly opinionated about this.

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This seems only tangentially related to the purpose of the PR.

Also, I'm not sure I agree, I think you would want to do:

RepairValidationResults repairResults;
    try (RecordRepairValidateRunner repairRunner = builder.buildRepairRunner(false)) {
        repairResults = repairRunner.run().join();
        // handle results
    }

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Reverted to previous behavior. Also, since CloseException is now declared on the signature of the method, I moved the exception to its own top level class.

}
}

@Nonnull
Expand Down
21 changes: 2 additions & 19 deletions ...oundationdb/record/provider/foundationdb/runners/throttled/ThrottledRetryingIterator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,30 +149,13 @@ public CompletableFuture<Void> iterateAll(final FDBRecordStore.Builder storeBuil
}

@Override
public void close() {
public void close() throws MoreAsyncUtil.CloseException {
if (closed) {
return;
}
closed = true;
// Ensure we call both close() methods, capturing all exceptions
RuntimeException caught = null;
try {
futureManager.close();
} catch (RuntimeException e) {
caught = e;
}
try {
transactionalRunner.close();
} catch (RuntimeException e) {
if (caught != null) {
caught.addSuppressed(e);
} else {
caught = e;
}
}
if (caught != null) {
throw caught;
}
MoreAsyncUtil.closeAll(futureManager, transactionalRunner);
}

/**
Expand Down