void resetUnconsumedOffsets() {
            Map<TopicPartition, Long> smallestUnconsumedOffsets = new HashMap<>();
            while (recordIter.hasNext()) {
                ConsumerRecord<byte[], byte[]> record = recordIter.next();
                TopicPartition tp = new TopicPartition(record.topic(), record.partition());
                // avoid auto-committing offsets which haven't been consumed
                smallestUnconsumedOffsets.putIfAbsent(tp, record.offset());
            }
            smallestUnconsumedOffsets.forEach(consumer::seek);
        }
Consumer that dumps messages to standard out.
tools/src/main/java/org/apache/kafka/tools/consumer/ConsoleConsumer.java