private ConsumerRecords<byte[], byte[]> pollRequests(final Duration pollTime) {
        ConsumerRecords<byte[], byte[]> records = ConsumerRecords.empty();

        lastPollMs = now;

        try {
            records = mainConsumer.poll(pollTime);
        } catch (final InvalidOffsetException e) {
            resetOffsets(e.partitions(), e);
        }

        return records;
    }
Get the next batch of records by polling.

@param pollTime how long to block in Consumer#poll
@return Next batch of records or null if no records available.
@throws TaskMigratedException if the task producer got fenced (EOS only)
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java