Copilot commented on code in PR #4881:
URL: https://github.com/apache/bookkeeper/pull/4881#discussion_r3974619108
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerOpenOp.java:
##########
@@ -310,6 +292,57 @@ public void readLastConfirmedComplete(int rc,
}
}
+ /**
+ * Callback completing the open once recovery is done, run on the handle's
thread. Without an ordering key
+ * this is the ledger-id keyed {@link OrderedGenericCallback}; with one,
the completion is submitted to the
+ * handle's executor, which is the thread selected by that key.
+ */
+ private GenericCallback<Void> recoveryCallback(boolean watchImmediately) {
+ if (orderingKey == null) {
+ return new OrderedGenericCallback<Void>(bk.getMainWorkerPool(),
ledgerId) {
+ @Override
+ public void safeOperationComplete(int rc, Void result) {
+ recoveryComplete(rc, watchImmediately);
+ }
+
+ @Override
+ public String toString() {
+ return String.format("Recover(%d)", ledgerId);
+ }
+ };
+ }
+ return (rc, result) -> {
+ try {
+ lh.executeOrdered(() -> recoveryComplete(rc,
watchImmediately));
+ } catch (RejectedExecutionException ree) {
+ log.warn().exception(ree).log("Failed to submit recovery
completion callback");
+ }
+ };
+ }
+
Review Comment:
If `lh.executeOrdered(...)` throws `RejectedExecutionException`, the
exception is only logged and the open operation is never completed (no
`openComplete(...)` path), which can hang callers waiting on the open
future/callback. Handle this by completing the open with an error (e.g.,
`InterruptedException`/client-closed equivalent) or by falling back to
completing on an alternative executor/thread so the open always terminates.
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java:
##########
@@ -2512,6 +2535,23 @@ void executeOrdered(Runnable runnable) throws
RejectedExecutionException {
executor.execute(runnable);
}
+ /**
+ * Run the task in the thread pinned to the ledger, exposing its result as
a future.
+ * @param task
+ * @throws RejectedExecutionException
+ */
+ <T> ListenableFuture<T> submitOrdered(Callable<T> task) throws
RejectedExecutionException {
+ SettableFuture<T> future = SettableFuture.create();
+ executeOrdered(() -> {
+ try {
+ future.set(task.call());
+ } catch (Throwable t) {
+ future.setException(t);
+ }
+ });
+ return future;
+ }
Review Comment:
The new `submitOrdered` Javadoc has incomplete `@param` and `@throws` tags.
Please document what `task` is expected to do and clarify when
`RejectedExecutionException` can be thrown (e.g., handle closed / executor
shutdown) to make the new API easier to use correctly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
