protected int maxNumPartitions(final Cluster metadata, final Set<String> topics) {
        int maxNumPartitions = 0;
        for (final String topic : topics) {
            final List<PartitionInfo> partitions = metadata.partitionsForTopic(topic);
            if (partitions.isEmpty()) {
                log.error("Empty partitions for topic {}", topic);
                throw new RuntimeException("Empty partitions for topic " + topic);
            }

            final int numPartitions = partitions.size();
            if (numPartitions > maxNumPartitions) {
                maxNumPartitions = numPartitions;
            }
        }
        return maxNumPartitions;
    }
@throws StreamsException if no metadata can be received for a topic
streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGrouper.java