public void consumeStrings(String topicName,int count,long timeout,TimeUnit unit,Runnable completion,BiPredicate<String,String> consumer){
  AtomicLong readCounter=new AtomicLong();
  consumeStrings(continueIfNotExpired(null,timeout,unit),completion,Collections.singleton(topicName),null);
}

public void consumeStrings ( String topicName , int count , long timeout , TimeUnit unit , Runnable completion , BiPredicate < String , String > consumer ) { AtomicLong readCounter = new AtomicLong ( ) ; consumeStrings ( continueIfNotExpired ( null , timeout , unit ) , completion , Collections . singleton ( topicName ) , null ) ; }
asynchronously consume all messages on the given topic from the cluster .
