private void addToResetList(final TopicPartition partition, final Set<TopicPartition> partitions, final String logMessage, final String resetPolicy, final Set<String> loggedTopics) {
        final String topic = partition.topic();
        if (loggedTopics.add(topic)) {
            log.info(logMessage, topic, resetPolicy);
        }
        partitions.add(partition);
    }
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