code: public static boolean seekToBeginning(KafkaConsumer<?, ?> consumer, String topic) {
          boolean result = false;
          synchronized (consumer) {
              Set<TopicPartition> topicParts = consumer.assignment();
              if (topicParts != null) {
                  for (TopicPartition tp : topicParts) {
                      if (StringUtils.equals(topic, tp.topic())) {
                          consumer.seekToBeginning(Arrays.asList(tp));
                          // we want to seek as soon as possible
                          // since seekToEnd evaluates lazily, invoke position()
                          // so
                          // that seeking will be committed.
                          consumer.position(tp);
                          result = true;
                      }
                  }
                  if (result) {
                      consumer.commitSync();
                  }
              }
          }
          return result;
      }
docs: Seeks the consumer's cursor to the beginning of a topic.
  
  <p>
  This method only set cursors of topic's partitions that are assigned to
  the consumer!
  </p>
  
  @param consumer
  @param topic
  @return {@code true} if the consumer has subscribed to the specified
  topic, {@code false} otherwise.