protected void pollOnce(long maxTimeoutMs) {
        try {
            drainGeneratedRequests();
            long now = time.milliseconds();
            final long timeout = sendRequests(now, maxTimeoutMs);
            networkClient.poll(timeout, now);
            now = time.milliseconds();
            checkDisconnects(now);
            failExpiredRequests(now);
            unsentRequests.clean();
        } catch (FatalExitError fee) {
            throw fee;
        } catch (Throwable t) {
            if (t instanceof DisconnectException && !networkClient.active()) {
                // DisconnectException is expected when NetworkClient#initiateClose is called
                return;
            }
            if (t instanceof InterruptedException && !isRunning()) {
                // InterruptedException is expected when shutting down. Throw the error to ShutdownableThread to handle
                throw t;
            }
            log.error("unhandled exception caught in InterBrokerSendThread", t);
            // rethrow any unhandled exceptions as FatalExitError so the JVM will be terminated
            // as we will be in an unknown state with potentially some requests dropped and not
            // being able to make progress. Known and expected Errors should have been appropriately
            // dealt with already.
            throw new FatalExitError();
        }
    }
@Test
    public void testDisconnectWithoutShutdownShouldCauseException() {
        DisconnectException de = new DisconnectException();
        when(networkClient.poll(anyLong(), anyLong())).thenThrow(de);
        when(networkClient.active()).thenReturn(true);

        AtomicReference<Throwable> throwable = new AtomicReference<>();
        final InterBrokerSendThread thread =
            new TestInterBrokerSendThread(networkClient, throwable::getAndSet);
        thread.pollOnce(100);

        verify(networkClient).poll(anyLong(), anyLong());
        verify(networkClient).active();
        verifyNoMoreInteractions(networkClient);

        Throwable thrown = throwable.get();
        assertNotNull(thrown);
        assertInstanceOf(FatalExitError.class, thrown);
    }