When managedLedgerNewEntriesCheckDelayInMillis=0, avoid the context switch
### Motivation
When `managedLedgerNewEntriesCheckDelayInMillis=0` is configured in broker.conf, in order to reduce the worst case for end-to-end latency, we should also avoid the context switch and paying the latency price for it.
@@ -738,8 +738,21 @@ public class ManagedCursorImpl implements ManagedCursor {
log.debug("[{}] [{}] Deferring retry of read at position {}", ledger.getName(), name, op.readPosition);
}
- // Check again for new entries again in 10ms, then if still no entries are available register to be notified
- ledger.getScheduledExecutor().schedule(safeRun(() -> {
+ // Check again for new entries after the configured time, then if still no entries are available register
+ // to be notified
+ if (config.getNewEntriesCheckDelayInMillis() > 0) {
+ ledger.getScheduledExecutor()
+ .schedule(() -> checkForNewEntries(op, callback, ctx),
+ config.getNewEntriesCheckDelayInMillis(), TimeUnit.MILLISECONDS);
+ } else {
+ // If there's no delay, check directly from the same thread
+ checkForNewEntries(op, callback, ctx);
+ }
+ }
+ }
+
+ private void checkForNewEntries(OpReadEntry op, ReadEntriesCallback callback, Object ctx) {
+ try {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Re-trying the read at position {}", ledger.getName(), name, op.readPosition);
}
@@ -783,7 +796,8 @@ public class ManagedCursorImpl implements ManagedCursor {
// If the managed ledger was indeed terminated, we need to notify the cursor
callback.readEntriesFailed(new NoMoreEntriesToReadException("Topic was terminated"), ctx);
}
- }), config.getNewEntriesCheckDelayInMillis(), TimeUnit.MILLISECONDS);
+ } catch (Throwable t) {
+ callback.readEntriesFailed(new ManagedLedgerException(t), ctx);
}
}
