@Override
        public BoxedUnit apply(Exception exception, ExponentialSleepStrategy.RetryLoop loop) {
          if (loop.sleepCount() < MAX_RETRIES_ON_EXCEPTION) {
            LOG.warn(
                String.format("Fetching system stream metadata for: %s threw an exception. Retrying.", streamNames),
                exception);
          } else {
            LOG.error(String.format("Fetching system stream metadata for: %s threw an exception.", streamNames),
                exception);
            loop.done();
            throw new SamzaException(exception);
          }

          return null;
        }
Given a set of stream names (topics), fetch metadata from Kafka for each
stream, and return a map from stream name to SystemStreamMetadata for
each stream. This method will return null for oldest and newest offsets
if a given SystemStreamPartition is empty. This method will block and
retry indefinitely until it gets a successful response from Kafka.

@param streamNames a set of strings of stream names/topics
@param retryBackoff retry backoff strategy
@return a map from topic to SystemStreamMetadata which has offsets for each partition
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemAdmin.java