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 ; }
Seeks the consumer s cursor to the beginning of a topic .