public void describeTopic(TopicCommandOptions opts) throws ExecutionException, InterruptedException {
            // If topicId is provided and not zero, will use topicId regardless of topic name
            Optional<Uuid> inputTopicId = opts.topicId()
                .map(Uuid::fromString).filter(uuid -> !uuid.equals(Uuid.ZERO_UUID));
            Boolean useTopicId = inputTopicId.isPresent();

            List<Uuid> topicIds;
            List<String> topics;
            if (useTopicId) {
                topicIds = getTopicIds(inputTopicId.get(), opts.excludeInternalTopics());
                topics = Collections.emptyList();
            } else {
                topicIds = Collections.emptyList();
                topics = getTopics(opts.topic(), opts.excludeInternalTopics());
            }

            // Only check topic name when topicId is not provided
            if (useTopicId) {
                ensureTopicIdExists(topicIds, inputTopicId.get(), !opts.ifExists());
            } else {
                ensureTopicExists(topics, opts.topic(), !opts.ifExists());
            }
            List<org.apache.kafka.clients.admin.TopicDescription> topicDescriptions = new ArrayList<>();

            if (!topicIds.isEmpty()) {
                Map<Uuid, org.apache.kafka.clients.admin.TopicDescription> descTopics =
                    adminClient.describeTopics(TopicCollection.ofTopicIds(topicIds)).allTopicIds().get();
                topicDescriptions = new ArrayList<>(descTopics.values());
            }

            if (!topics.isEmpty()) {
                Map<String, org.apache.kafka.clients.admin.TopicDescription> descTopics =
                    adminClient.describeTopics(TopicCollection.ofTopicNames(topics),
                        new DescribeTopicsOptions()
                            .partitionSizeLimitPerResponse(opts.partitionSizeLimitPerResponse().orElse(2000))).allTopicNames().get();
                topicDescriptions = new ArrayList<>(descTopics.values());
            }

            List<String> topicNames = topicDescriptions.stream()
                .map(org.apache.kafka.clients.admin.TopicDescription::name)
                .collect(Collectors.toList());
            Map<ConfigResource, KafkaFuture<Config>> allConfigs = adminClient.describeConfigs(
                topicNames.stream()
                    .map(name -> new ConfigResource(ConfigResource.Type.TOPIC, name))
                    .collect(Collectors.toList())
            ).values();
            List<Integer> liveBrokers = adminClient.describeCluster().nodes().get().stream()
                .map(Node::id)
                .collect(Collectors.toList());
            DescribeOptions describeOptions = new DescribeOptions(opts, new HashSet<>(liveBrokers));
            Set<TopicPartition> topicPartitions = topicDescriptions
                .stream()
                .flatMap(td -> td.partitions().stream()
                    .map(p -> new TopicPartition(td.name(), p.partition())))
                .collect(Collectors.toSet());
            Map<TopicPartition, PartitionReassignment> reassignments = listAllReassignments(topicPartitions);
            for (org.apache.kafka.clients.admin.TopicDescription td : topicDescriptions) {
                String topicName = td.name();
                Uuid topicId = td.topicId();
                Config config = allConfigs.get(new ConfigResource(ConfigResource.Type.TOPIC, topicName)).get();
                ArrayList<TopicPartitionInfo> sortedPartitions = new ArrayList<>(td.partitions());
                sortedPartitions.sort(Comparator.comparingInt(TopicPartitionInfo::partition));
                printDescribeConfig(opts, describeOptions, reassignments, td, topicName, topicId, config, sortedPartitions);
                printPartitionDescription(describeOptions, reassignments, td, topicName, config, sortedPartitions);
            }
        }
ensures topic existence and throws exception if topic doesn't exist

@param foundTopicIds        Topics that were found to match the requested topic id.
@param requestedTopicId     Id of the topic that was requested.
@param requireTopicIdExists Indicates if the topic needs to exist for the operation to be successful.
                            If set to true, the command will throw an exception if the topic with the
                            requested id does not exist.
tools/src/main/java/org/apache/kafka/tools/TopicCommand.java