@Override
    public PollResult poll(long currentTimeMs) {
        return pollInternal(
                prepareFetchRequests(),
                this::handleFetchSuccess,
                this::handleFetchFailure
        );
    }
@Test
    public void testFetcherCloseClosesFetchSessionsInBroker() {
        buildFetcher();

        assignFromUser(singleton(tp0));
        subscriptions.seek(tp0, 0);

        // normal fetch
        assertEquals(1, sendFetches());
        assertFalse(fetcher.hasCompletedFetches());

        final FetchResponse fetchResponse = fullFetchResponse(tidp0, records, Errors.NONE, 100L, 0);
        client.prepareResponse(fetchResponse);
        networkClientDelegate.poll(time.timer(0));
        assertTrue(fetcher.hasCompletedFetches());
        assertEquals(0, networkClientDelegate.pendingRequestCount());

        final ArgumentCaptor<NetworkClientDelegate.UnsentRequest> argument = ArgumentCaptor.forClass(NetworkClientDelegate.UnsentRequest.class);

        Timer timer = time.timer(Duration.ofSeconds(10));
        // NOTE: by design the FetchRequestManager doesn't perform network I/O internally. That means that calling
        // the close() method with a Timer will NOT send out the close session requests on close. The network
        // I/O logic is handled inside ConsumerNetworkThread.runAtClose, so we need to run that logic here.
        ConsumerNetworkThread.runAtClose(singletonList(Optional.of(fetcher)), networkClientDelegate);
        // the network is polled during the last state of clean up.
        networkClientDelegate.poll(time.timer(1));
        // validate that closing the fetcher has sent a request with final epoch. 2 requests are sent, one for the
        // normal fetch earlier and another for the finish fetch here.
        verify(networkClientDelegate, times(2)).doSend(argument.capture(), any(Long.class));
        NetworkClientDelegate.UnsentRequest lastUnsentRequest = argument.getValue();
        FetchRequest.Builder builder = (FetchRequest.Builder) lastUnsentRequest.requestBuilder();
        // session Id is the same
        assertEquals(fetchResponse.sessionId(), builder.metadata().sessionId());
        // contains final epoch
        assertEquals(FetchMetadata.FINAL_EPOCH, builder.metadata().epoch());  // final epoch indicates we want to close the session
        assertTrue(builder.fetchData().isEmpty()); // partition data should be empty
    }