override def describeTopic(opts: TopicCommandOptions): Unit = {
      val topics = getTopics(opts.topic, opts.excludeInternalTopics)
      ensureTopicExists(topics, opts.topic, !opts.ifExists)

      if (topics.nonEmpty) {
        val allConfigs = adminClient.describeConfigs(topics.map(new ConfigResource(Type.TOPIC, _)).asJavaCollection).values()
        val liveBrokers = adminClient.describeCluster().nodes().get().asScala.map(_.id())
        val topicDescriptions = adminClient.describeTopics(topics.asJavaCollection).all().get().values().asScala
        val describeOptions = new DescribeOptions(opts, liveBrokers.toSet)
        val topicPartitions = topicDescriptions
          .flatMap(td => td.partitions.iterator().asScala.map(p => new TopicPartition(td.name(), p.partition())))
          .toSet.asJava
        val reassignments = listAllReassignments(topicPartitions)

        if (opts.reportNonPreferredLeaderJson) {
          var nonPreferredLeader = Set.empty[TopicPartition]
          for (td <- topicDescriptions) {
            val topicName = td.name
            val config = allConfigs.get(new ConfigResource(Type.TOPIC, topicName)).get()
            val sortedPartitions = td.partitions.asScala.sortBy(_.partition)

            nonPreferredLeader = nonPreferredLeader.++(sortedPartitions.toSet.filter { tpInfo =>
              val reassignment = reassignments.get(new TopicPartition(topicName, tpInfo.partition))
              val partitionDesc = PartitionDescription(topicName, tpInfo, Some(config), markedForDeletion = false, reassignment)
              describeOptions.shouldPrintTopicPartition(partitionDesc)
            }.map(tp => new TopicPartition(topicName, tp.partition())))
          }
          println(formatAsJson(nonPreferredLeader))
        } else {
          for (td <- topicDescriptions) {
            val topicName = td.name
            val topicId = td.topicId()
            val config = allConfigs.get(new ConfigResource(Type.TOPIC, topicName)).get()
            val sortedPartitions = td.partitions.asScala.sortBy(_.partition)

            if (describeOptions.describeConfigs) {
              val hasNonDefault = config.entries().asScala.exists(!_.isDefault)
              if (!opts.reportOverriddenConfigs || hasNonDefault) {
                val numPartitions = td.partitions().size
                val firstPartition = td.partitions.iterator.next()
                val reassignment = reassignments.get(new TopicPartition(td.name, firstPartition.partition))
                val topicDesc = TopicDescription(topicName, topicId, numPartitions, getReplicationFactor(firstPartition, reassignment), config, markedForDeletion = false)
                topicDesc.printDescription()
     def isUnderPreferredReplicaPartitions         }
            }

            if (describeOptions.describePartitions) {
              for (partition <- sortedPartitions) {
                val reassignment = reassignments.get(new TopicPartition(td.name, partition.partition))
                val partitionDesc = PartitionDescription(topicName, partition, Some(config), markedForDeletion = false, reassignment)
                describeOptions.maybePrintPartitionDescription(partitionDesc)
              }
            }
          }
        }
      }
    }

    def shouldPrintTopicPartition(partitionDesc: PartitionDescription): Boolean = {
      describeConfigs ||
        shouldPrintUnderReplicatedPartitions(partitionDesc) ||
        shouldPrintUnavailablePartitions(partitionDesc) ||
        shouldPrintUnderMinIsrPartitions(partitionDesc) ||
        shouldPrintAtMinIsrPartitions(partitionDesc) ||
        shouldPrintNonPreferredLeader(partitionDesc)
    }

    def isNonPreferredLeader: Boolean = {
      hasLeader && !info.leader.equals(info.replicas.asScala.head)
    }

    private def shouldPrintUnderPreferredReplicaPartitionsshouldPrintNonPreferredLeader(partitionDescription: PartitionDescription): Boolean = {
      (opts.reportUnderPreferredReplicaPartitionsreportNonPreferredLeader || opts.reportNonPreferredLeaderJson) && partitionDescription.isUnderPreferredReplicaPartitionsisNonPreferredLeader
    }

    private val reportUnderPreferredReplicaPartitionsOptreportNonPreferredLeaderOpt = parser.accepts("undernon-preferred-leader",
      "if set when describing topics, only show partitions whose leader is not equal to the first replica-partitions in the replica list. Not supported with the --zookeeper option.")
    private val reportNonPreferredLeaderJsonOpt = parser.accepts("non-preferred-leader-json",
      "if set when describing topics, only show partitions whose leader is not equal to the first replica in the replica list and output as json format. Not supported with the --zookeeper option.")

    private val allReplicationReportOpts = Set(reportUnderReplicatedPartitionsOpt, reportUnderMinIsrPartitionsOpt, reportAtMinIsrPartitionsOpt, reportUnavailablePartitionsOpt, reportNonPreferredLeaderOpt, reportUnderPreferredReplicaPartitionsOptreportNonPreferredLeaderJsonOpt)

    def reportUnderPreferredReplicaPartitions reportNonPreferredLeader: Boolean = has(reportNonPreferredLeaderOpt)
    def reportNonPreferredLeaderJson: Boolean = has(reportUnderPreferredReplicaPartitionsOpt)(reportNonPreferredLeaderJsonOpt)

    private def formatAsJson(nonPreferredLeader: Set[TopicPartition]): String = {
      Json.encodeAsString(Map(
        "partitions" -> nonPreferredLeader.map { tp =>
          Map(
            "topic" -> tp.topic(),
            "partition" -> tp.partition()
          ).asJava
        }.asJava
      ).asJava)
    }

	# Some other minor changes omitted