public synchronized int sendFetches() {
        final Map<Node, FetchSessionHandler.FetchRequestData> fetchRequests = prepareFetchRequests();
        sendFetchesInternal(
                fetchRequests,
                (fetchTarget, data, clientResponse) -> {
                    synchronized (Fetcher.this) {
                        handleFetchSuccess(fetchTarget, data, clientResponse);
                    }
                },
                (fetchTarget, data, error) -> {
                    synchronized (Fetcher.this) {
                        handleFetchFailure(fetchTarget, data, error);
                    }
                });
        return fetchRequests.size();
    }
@Test
    public void testFetcherSessionEpochUpdate() throws Exception {
        buildFetcher(2);

        MetadataResponse initialMetadataResponse = RequestTestUtils.metadataUpdateWithIds(1, singletonMap(topicName, 1), topicIds);
        client.updateMetadata(initialMetadataResponse);
        assignFromUser(Collections.singleton(tp0));
        subscriptions.seek(tp0, 0L);

        AtomicInteger fetchesRemaining = new AtomicInteger(1000);
        executorService = Executors.newSingleThreadExecutor();
        Future<?> future = executorService.submit(() -> {
            long nextOffset = 0;
            long nextEpoch = 0;
            while (fetchesRemaining.get() > 0) {
                synchronized (consumerClient) {
                    if (!client.requests().isEmpty()) {
                        ClientRequest request = client.requests().peek();
                        FetchRequest fetchRequest = (FetchRequest) request.requestBuilder().build();
                        int epoch = fetchRequest.metadata().epoch();
                        assertTrue(epoch == 0 || epoch == nextEpoch,
                            String.format("Unexpected epoch expected %d got %d", nextEpoch, epoch));
                        nextEpoch++;
                        LinkedHashMap<TopicIdPartition, FetchResponseData.PartitionData> responseMap = new LinkedHashMap<>();
                        responseMap.put(tidp0, new FetchResponseData.PartitionData()
                                .setPartitionIndex(tp0.partition())
                                .setHighWatermark(nextOffset + 2)
                                .setLastStableOffset(nextOffset + 2)
                                .setLogStartOffset(0)
                                .setRecords(buildRecords(nextOffset, 2, nextOffset)));
                        nextOffset += 2;
                        client.respondToRequest(request, FetchResponse.of(Errors.NONE, 0, 123, responseMap));
                        consumerClient.poll(time.timer(0));
                    }
                }
            }
            return fetchesRemaining.get();
        });
        long nextFetchOffset = 0;
        while (fetchesRemaining.get() > 0 && !future.isDone()) {
            if (sendFetches() == 1) {
                synchronized (consumerClient) {
                    consumerClient.poll(time.timer(0));
                }
            }
            if (fetcher.hasCompletedFetches()) {
                Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> fetchedRecords = fetchRecords();
                if (!fetchedRecords.isEmpty()) {
                    fetchesRemaining.decrementAndGet();
                    List<ConsumerRecord<byte[], byte[]>> records = fetchedRecords.get(tp0);
                    assertEquals(2, records.size());
                    assertEquals(nextFetchOffset, records.get(0).offset());
                    assertEquals(nextFetchOffset + 1, records.get(1).offset());
                    nextFetchOffset += 2;
                }
                assertTrue(fetchRecords().isEmpty());
            }
        }
        assertEquals(0, future.get());
    }